convert kmeans_incr threshold

[actl.git] / app-l3_lb / l3_lb.c

/*
 * Copyright (c) 2016 Mohamed Aslan <maslan@sce.carleton.ca>
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <getopt.h>
#include <err.h>
#include <time.h>

#include <libof.h>
#include <of10.h>
#include <hashtab.h>

#include <net/if_arp.h>
#include <netinet/in.h>
#include <netinet/if_ether.h>
#include <arpa/inet.h>
#include <netinet/ip.h>
#include <sys/queue.h>

#include <actl.h>
#include "l3_lb.h"

#ifdef __linux__
#include <netinet/ether.h>
#endif

/*
        when a packet arrives:
        (1) learn (in port and src mac)
        (2) check whether it's an arp or ip packet
           a. in case of arp:
              i. if dst is vserver:
                 fake arp response
              ii. otherwise:
                 flood
           b. in case of ip:
              i. if dst is vserver:
                 loadbalance
              ii. otherwise:
                 flood or forward
*/

#define DPID(dp) (be64toh(((struct of10_switch_features *)((dp)->sw_features))->datapath_id))

/* mactable's key entry */
struct entry {
        uint64_t        dpid;   
        uint8_t         mac_addr[ETHER_ADDR_LEN];
};

static SLIST_HEAD(, dp_ptr) dataplane_list;
struct dp_ptr {
        struct of_dataplane     *dp;
        SLIST_ENTRY(dp_ptr)      next;
};

static SLIST_HEAD(, server) server_list;
struct server {
        struct in_addr          ip_addr;
        struct ether_addr       mac_addr;
        int                     ip_resolved;
        uint64_t                load;
        uint64_t                load_bytes;
        time_t                  load_ts;
        SLIST_ENTRY(server)     next;
};

static struct actl_ctx          *ctx;
static struct hashtab            mactable;
static struct in_addr            vserver_ip;    /* virtual server's IP */
static struct ether_addr         vserver_mac;   /* virtual server's MAC */
static int                       n_servers = 0;
static int                       auto_arp = 0;
static uint16_t                  ttl_idle = DEFAULT_TIMEOUT;
static uint16_t                  ttl_hard = 0;

static struct option longopts[] = { 
        { "ip_addr",            required_argument,      NULL,   'i' }, 
        { "mac_addr",           required_argument,      NULL,   'm' }, 
        { "servers",            required_argument,      NULL,   's' }, 
        { "arp_servers",        required_argument,      NULL,   'a'},
        { "poll",               required_argument,      NULL,   'p'},
        { "kpi",                required_argument,      NULL,   'k'},
        { "ttl_idle",           required_argument,      NULL,   't'},
        { "ttl_hard",           required_argument,      NULL,   'h'},
        { NULL,                 0,                      NULL,   0 } 
};



static void
learn(struct of_dataplane *dp, struct of10_packet_in *p_in)
{
        struct ether_header     *eh;
        struct entry             e;
        struct ether_addr        broadcast_addr;

        eh = (struct ether_header *)(p_in->data);
        memset(&broadcast_addr, 0xff, ETHER_ADDR_LEN);

        /* skip the load-balacing address from the learning process */
        if (!memcmp(eh->ether_shost, &vserver_mac, ETHER_ADDR_LEN))
                return;
        /* skip broadcast address */
        if (!memcmp(eh->ether_shost, &broadcast_addr, ETHER_ADDR_LEN))
                return;

        /* L2 learning (update mactable) */
        e.dpid = be64toh(((struct of10_switch_features *)(dp->sw_features))->datapath_id);
        memcpy(e.mac_addr, eh->ether_shost, ETHER_ADDR_LEN);
        hashtab_put(&mactable, &e, sizeof(struct entry), &p_in->in_port, sizeof(uint16_t));
#ifdef DEBUG
        printf("[dpid=0x%016llx] learned that %s is at port %hu.\n", DPID(dp), ether_ntoa(eh->ether_shost), ntohs(p_in->in_port));
#endif
}

static int
lookup(struct of_dataplane *dp, struct ether_addr *ea, uint16_t *port)
{
        uint16_t        *p;
        size_t           len;
        struct entry     e;

        e.dpid = be64toh(((struct of10_switch_features *)(dp->sw_features))->datapath_id);
        memcpy(e.mac_addr, ea, ETHER_ADDR_LEN);
#ifdef DEBUG
        printf("[dpid=0x%016llx] looking up %s.\n", DPID(dp), ether_ntoa(ea));
#endif
        if (hashtab_get(&mactable, &e, sizeof(struct entry), &p, &len)) {
                *port = *p;
                return 1;
        }
        return 0;
}

static void
flood(struct of_controller *ctl, struct of_dataplane *dp, struct of10_packet_in *p_in)
{
        struct of10_packet_out          *msg_out;
        struct of10_action_output       *action;

        msg_out = (struct of10_packet_out *)malloc(sizeof(struct of10_packet_out) + sizeof(struct of10_action_output));
        if (msg_out == NULL)
                return;
        msg_out->hdr.version = OFP_VERSION_10;
        msg_out->hdr.type = OFPT10_PACKET_OUT;
        msg_out->hdr.length = htons(sizeof(struct of10_packet_out) + sizeof(struct of10_action_output));
        msg_out->hdr.xid = 0;
        msg_out->buffer_id = p_in->buffer_id;
        msg_out->in_port = p_in->in_port;
        msg_out->actions_len = htons(sizeof(struct of10_action_output));
        action = (struct of10_action_output *)(msg_out->actions);
        action->type = htons(OFPAT10_OUTPUT);
        action->len = htons(sizeof(struct of10_action_output));
        action->port = htons(OFPP10_FLOOD);
        action->max_len = 0;
        ctl->send(ctl, dp, (struct ofp_header *)msg_out);
        free(msg_out);
#ifdef DEBUG
        printf("packet flooded.\n");
#endif
}

static void
forward_install(struct of_controller *ctl, struct of_dataplane *dp, struct of10_packet_in *p_in, uint16_t out_port)
{
        struct ether_header             *eh;
        struct of10_flow_mod            *msg_mod;
        struct of10_action_output       *action;

        eh = (struct ether_header *)(p_in->data);
        msg_mod = (struct of10_flow_mod *)malloc(sizeof(struct of10_flow_mod) + sizeof(struct of10_action_output));
        if (msg_mod == NULL)
                return;
        /* header */
        msg_mod->hdr.version = OFP_VERSION_10;
        msg_mod->hdr.type = OFPT10_FLOW_MOD;
        msg_mod->hdr.length = htons(sizeof(struct of10_flow_mod) + sizeof(struct of10_action_output));
        msg_mod->hdr.xid = 0;
        /* match */
        msg_mod->match.wildcards = htonl(OFPFW10_ALL & ~(OFPFW10_DL_DST));
        memcpy(msg_mod->match.dl_dst, eh->ether_dhost, OFP10_ETH_ALEN);
        /* others */
        msg_mod->cookie = 0;
        msg_mod->command = htons(OFPFC10_ADD);
        msg_mod->idle_timeout = htons(ttl_idle);
        msg_mod->hard_timeout = htons(ttl_hard);
        msg_mod->priority = htons(0xffff);
        msg_mod->buffer_id = p_in->buffer_id;
        msg_mod->out_port = out_port;   /* already in net byte order */
        msg_mod->flags = 0;
        /* action */
        action = (struct of10_action_output *)(msg_mod->actions);
        action->type = htons(OFPAT10_OUTPUT);
        action->len = htons(sizeof(struct of10_action_output));
        action->port = out_port;        /* already in net byte order */
        action->max_len = 0;
        ctl->send(ctl, dp, (struct ofp_header *)msg_mod);
        free(msg_mod);
}

static void
rewrite_install(struct of_controller *ctl, struct of_dataplane *dp, struct of10_packet_in *p_in, struct server *srv, uint16_t out_port)
{
        struct ether_header             *eh;
        struct of10_flow_mod            *msg_mod;
        struct of10_action_nw_addr      *action0;
        struct of10_action_dl_addr      *action1;
        struct of10_action_output       *action2;

        eh = (struct ether_header *)(p_in->data);
        msg_mod = (struct of10_flow_mod *)malloc(sizeof(struct of10_flow_mod) + 3 * sizeof(struct of10_action_output));
        if (msg_mod == NULL)
                return;

        /* source -> server */
        /* header */
        msg_mod->hdr.version = OFP_VERSION_10;
        msg_mod->hdr.type = OFPT10_FLOW_MOD;
        msg_mod->hdr.length = htons(sizeof(struct of10_flow_mod) + sizeof(struct of10_action_nw_addr) +
                sizeof(struct of10_action_dl_addr) + sizeof(struct of10_action_output));
        msg_mod->hdr.xid = 0;
        /* match */
        msg_mod->match.wildcards = htonl(OFPFW10_ALL & ~(OFPFW10_DL_SRC | OFPFW10_DL_DST));
        memcpy(msg_mod->match.dl_src, eh->ether_shost, OFP10_ETH_ALEN);
        memcpy(msg_mod->match.dl_dst, eh->ether_dhost, OFP10_ETH_ALEN);
        /* others */
        msg_mod->cookie = 0;
        msg_mod->command = htons(OFPFC10_ADD);
        msg_mod->idle_timeout = htons(ttl_idle);
        msg_mod->hard_timeout = htons(ttl_hard);
        msg_mod->priority = htons(0xffff);
        msg_mod->buffer_id = p_in->buffer_id;
        msg_mod->out_port = out_port;   /* already in net byte order */
        msg_mod->flags = 0;
        /* actions */
        /* rewrite ip address */
        action0 = (struct of10_action_nw_addr *)(msg_mod->actions);
        action0->type = htons(OFPAT10_SET_NW_DST);
        action0->len = htons(sizeof(struct of10_action_nw_addr));
        action0->nw_addr = srv->ip_addr.s_addr; /* already in net byte order */
        /* rewrite ether address */
        action1 = (struct of10_action_dl_addr *)(++action0);
        action1->type = htons(OFPAT10_SET_DL_DST);
        action1->len = htons(sizeof(struct of10_action_dl_addr));
        memcpy(&action1->dl_addr, &srv->mac_addr, OFP10_ETH_ALEN);
        /* forward */
        action2 = (struct of10_action_output *)(++action1);
        action2->type = htons(OFPAT10_OUTPUT);
        action2->len = htons(sizeof(struct of10_action_output));
        action2->port = out_port;       /* already in net byte order */
        action2->max_len = 0;
        /* send flow mod message */
        ctl->send(ctl, dp, (struct ofp_header *)msg_mod);
#ifdef DEBUG
        printf("[LB] rewrite and install (1).\n");
#endif

        /* server -> source */
        /* reuse msg_mod */
        memcpy(msg_mod->match.dl_src, &srv->mac_addr, OFP10_ETH_ALEN);
        memcpy(msg_mod->match.dl_dst, eh->ether_shost, OFP10_ETH_ALEN);
        msg_mod->buffer_id = htonl(-1);
        /* rewrite ip address */
        action0 = (struct of10_action_nw_addr *)(msg_mod->actions);
        action0->type = htons(OFPAT10_SET_NW_SRC);
        action0->nw_addr = vserver_ip.s_addr;   /* already in net byte order */
        /* rewrite ether address */
        action1 = (struct of10_action_dl_addr *)(++action0);
        action1->type = htons(OFPAT10_SET_DL_SRC);
        memcpy(&action1->dl_addr, &vserver_mac, OFP10_ETH_ALEN);
        /* forward */
        action2 = (struct of10_action_output *)(++action1);
        action2->port = p_in->in_port;
        /* send flow mod message */
        ctl->send(ctl, dp, (struct ofp_header *)msg_mod);
        free(msg_mod);
#ifdef DEBUG
        printf("[LB] rewrite and install (2).\n");
#endif
}

static void
loadbalance(struct of_controller *ctl, struct of_dataplane *dp, struct of10_packet_in *p_in)
{
        uint64_t         least_load;
        uint16_t         llsrv_port;
        struct server   *srv, *llsrv = NULL;

        if (!n_servers)
                return; 

        /* find least-loaded server */
        srv = SLIST_FIRST(&server_list);
        least_load = srv->load;
        SLIST_FOREACH(srv, &server_list, next) {
                printf("[LB] server %s -> load is %lld bytes/sec (status is %d).\n", inet_ntoa(srv->ip_addr), srv->load, srv->ip_resolved);
                if (!srv->ip_resolved)
                        continue;       /* next server */
                if ((llsrv == NULL) || (srv->load < least_load)) {
                        least_load = srv->load;
                        llsrv = srv;
                }
        }
        if (llsrv == NULL)
                return;         /* TODO: drop packet */
done:
        printf("[LB] least-loaded server is %s (with load = %lld bytes/sec).\n", inet_ntoa(llsrv->ip_addr), llsrv->load);
        if (!lookup(dp, &llsrv->mac_addr, &llsrv_port))
                llsrv_port = htons(OFPP10_FLOOD);
        printf("[LB] least-loaded server's port is %hu.\n", ntohs(llsrv_port));
        rewrite_install(ctl, dp, p_in, llsrv, llsrv_port);
}

static void
learn_servers(struct ether_arp *arp)
{
        struct ether_addr               *arp_sha;
        struct in_addr                  *arp_spa;
        struct server                   *srv;

        arp_sha = (struct ether_addr *)arp->arp_sha;
        arp_spa = (struct in_addr *)arp->arp_spa;

        /* check it's one of the servers (L3 learning) */
        SLIST_FOREACH(srv, &server_list, next) {
                if (arp_spa->s_addr == srv->ip_addr.s_addr) {
                        memcpy(&srv->mac_addr, arp_sha, ETHER_ADDR_LEN);
                        srv->ip_resolved = 1;
                        printf("learned (L3) that server %s is at %s.\n", inet_ntoa(srv->ip_addr), ether_ntoa(&srv->mac_addr));
                        break;
                }
        }
}

static void
handle_arp(struct of_controller *ctl, struct of_dataplane *dp, struct of10_packet_in *ph_in)
{
        struct ether_header             *eh_in, *eh_out;
        struct ether_arp                *arp_in, *arp_out;
        struct of10_packet_out          *msg_out;
        struct of10_action_output       *action;
        struct in_addr                  *arp_in_tpa;
        uint8_t                         *p;

        eh_in = (struct ether_header *)(ph_in->data);
        arp_in = (struct  ether_arp *)(ph_in->data + ETHER_HDR_LEN);

        /* check if it's valid or not */
        if (!(arp_in->ea_hdr.ar_hrd == htons(ARPHRD_ETHER)) ||
                !(arp_in->ea_hdr.ar_pro == htons(ETHERTYPE_IP)))
                return;
        /* check if it's an arp reply */
        if (arp_in->ea_hdr.ar_op == htons(ARPOP_REPLY)) {
                printf("[ARP] %u.%u.%u.%u is at %02x:%02x:%02x:%02x:%02x:%02x\n",
                        arp_in->arp_spa[0], arp_in->arp_spa[1], arp_in->arp_spa[2], arp_in->arp_spa[3],
                        arp_in->arp_sha[0], arp_in->arp_sha[1], arp_in->arp_sha[2], arp_in->arp_sha[3],
                        arp_in->arp_sha[4], arp_in->arp_sha[5]);
                learn_servers(arp_in);  /* L3 learn about the servers */
                flood(ctl, dp, ph_in);  /* XXX: lookup() */
                return;
        }

        /* now, ensure it's an arp request */
        if (arp_in->ea_hdr.ar_op != htons(ARPOP_REQUEST))
                return;
        /* okay */
        printf("[ARP] who has %u.%u.%u.%u tell %u.%u.%u.%u\n",
                arp_in->arp_tpa[0], arp_in->arp_tpa[1], arp_in->arp_tpa[2], arp_in->arp_tpa[3],
                arp_in->arp_spa[0], arp_in->arp_spa[1], arp_in->arp_spa[2], arp_in->arp_spa[3]);

        /* check if it's destined to vserver or not */
        arp_in_tpa = (struct in_addr *)arp_in->arp_tpa;
        if (arp_in_tpa->s_addr != vserver_ip.s_addr) {
                printf("[ARP] arp is not for a load balanced host.\n");
                flood(ctl, dp, ph_in);
                return;
        }
        printf("[ARP] arp is for a load balanced host.\n");

        /* fake arp response */
        msg_out = (struct of10_packet_out *)
                malloc(sizeof(struct of10_packet_out) + sizeof(struct of10_action_output)
                + ETHER_HDR_LEN + sizeof(struct  ether_arp));
        if (msg_out == NULL)
                return;
        /*
        eh_out = (struct ether_header *)(msg_out + sizeof(struct of10_packet_out) +
                sizeof(struct of10_action_output));
        arp_out = (struct ether_arp *)(msg_out + sizeof(struct of10_packet_out) +
                sizeof(struct of10_action_output) + ETHER_HDR_LEN);
        */

        p = (uint8_t *)(msg_out) + sizeof(struct of10_packet_out) +
                sizeof(struct of10_action_output);
        eh_out = (struct ether_header *)p;
        p += ETHER_HDR_LEN;
        arp_out = (struct ether_arp *)p;

        /* ether_arp */
        arp_out->ea_hdr.ar_hrd = htons(ARPHRD_ETHER);
        arp_out->ea_hdr.ar_pro = htons(ETHERTYPE_IP);
        arp_out->ea_hdr.ar_op = htons(ARPOP_REPLY);
        arp_out->ea_hdr.ar_hln = arp_in->ea_hdr.ar_hln;
        arp_out->ea_hdr.ar_pln = arp_in->ea_hdr.ar_pln;
        memcpy(arp_out->arp_tha, arp_in->arp_sha, sizeof(arp_out->arp_tha));
        memcpy(arp_out->arp_tpa, arp_in->arp_spa, sizeof(arp_out->arp_tpa));
        memcpy(arp_out->arp_sha, &vserver_mac, sizeof(arp_out->arp_sha));
        memcpy(arp_out->arp_spa, &vserver_ip, sizeof(arp_out->arp_spa));
        /* ether_header */
        eh_out->ether_type = ntohs(ETHERTYPE_ARP);
        memcpy(eh_out->ether_dhost, eh_in->ether_shost, sizeof(eh_out->ether_dhost));
        memcpy(eh_out->ether_shost, &vserver_mac, sizeof(eh_out->ether_shost));
        /* packet_out */        
        msg_out->hdr.version = OFP_VERSION_10;
        msg_out->hdr.type = OFPT10_PACKET_OUT;
        msg_out->hdr.length = htons(sizeof(struct of10_packet_out) + sizeof(struct of10_action_output) + ETHER_HDR_LEN + sizeof(struct  ether_arp));
        msg_out->hdr.xid = 0;
        msg_out->buffer_id = htonl(-1);
        msg_out->in_port = ph_in->in_port;
        msg_out->actions_len = htons(sizeof(struct of10_action_output));
        action = (struct of10_action_output *)(msg_out->actions);
        action->type = htons(OFPAT10_OUTPUT);
        action->len = htons(sizeof(struct of10_action_output));
        action->port = htons(OFPP10_IN_PORT);
        action->max_len = 0;
        ctl->send(ctl, dp, (struct ofp_header *)msg_out);
        free(msg_out);
}

static void
handle_ip(struct of_controller *ctl, struct of_dataplane *dp, struct of10_packet_in *p_in)
{
        struct ether_header             *eh_in;
        struct ip                       *ip_in;
        uint16_t                         out_port;

        eh_in = (struct ether_header *)(p_in->data);
        ip_in = (struct ether_arp *)(p_in->data + ETHER_HDR_LEN);

        char *x = strdup(inet_ntoa(ip_in->ip_src));
        char *y = strdup(inet_ntoa(ip_in->ip_dst));
        printf("[IP] from %s to %s\n", x, y);
        free(x);
        free(y);

        /* check if it's destined to vserver or not */
        if (ip_in->ip_dst.s_addr == vserver_ip.s_addr) {
                printf("[IP] packet is for a load balanced host.\n");
                loadbalance(ctl, dp, p_in);
                return;
        }

        if (lookup(dp, (struct ether_addr *)eh_in->ether_dhost, &out_port)) {
                printf("[IP] forward (port = %hu) and install flow rule.\n", ntohs(out_port));
                forward_install(ctl, dp, p_in, out_port);
        }
        else{
                printf("[IP] flood (%s not found).\n", ether_ntoa(eh_in->ether_dhost));
                flood(ctl, dp, p_in);
        }
}

static void
handle_stats(struct of_controller *ctl, struct of_dataplane *dp, struct ofp_header *msg)
{
        int                               i, n;
        struct of10_stats_reply          *sts;
        struct of10_port_stats           *psts;
        struct server                    *srv;
        struct host_info                **hosts;
        time_t                            now;
        char                             *loadstr;


        sts = (struct of10_stats_reply *)msg;
#ifdef DEBUG
        printf("[DEBUG] stats received type=0x%x, size=0x%x.\n", ntohs(sts->type), ntohs(sts->hdr.length));
#endif
        /* process only port stats for now */
        if (ntohs(sts->type) != OFPST10_PORT)
                return;
        psts = (struct of10_flow_stats *)sts->body;
        if (psts == NULL)
                return;
#ifdef DEBUG
        printf("[DEBUG] dpid=0x%016llx,port=%u load is now %llu bytes.\n", DPID(dp), ntohs(psts->port_no), be64toh(psts->tx_bytes));
#endif
        /* read static topology information */
        ctx->topo(ctx, &hosts, &n);
        for (i = 0 ; i < n ; i++) {
                if ((DPID(dp) == hosts[i]->dpid) && (ntohs(psts->port_no) == hosts[i]->port)) {
                        SLIST_FOREACH(srv, &server_list, next) {
                                if (srv->ip_addr.s_addr == hosts[i]->ip_addr.s_addr) {
                                        now = time(NULL);
                                        srv->load = (uint64_t)(be64toh(psts->tx_bytes) - srv->load_bytes) / (now - srv->load_ts);
                                        srv->load_bytes = (uint64_t)be64toh(psts->tx_bytes);
                                        srv->load_ts = now;
                                        printf("[INFO] server %s load is now %llu bytes/sec.\n", inet_ntoa(srv->ip_addr), srv->load);
                                        break;
                                }
                        }
                }
        }

        (void)asprintf(&loadstr, "%llu", srv->load);
        if (!loadstr)
                errx(1, "failed to allocate memory");
        n = ctx->put(ctx, inet_ntoa(srv->ip_addr), loadstr);
        if (!n)
                printf("[INSERT] put(%s, %s) failed.\n", inet_ntoa(srv->ip_addr), loadstr);
        else
                printf("[INSERT] put(%s, %s) succeeded.\n", inet_ntoa(srv->ip_addr), loadstr);
        free(loadstr);
}

static void
handle_packet_in(struct of_controller *ctl, struct of_dataplane *dp, struct ofp_header *msg_in)
{
        struct ether_header             *eh;
        struct of10_packet_in           *p_in;


        p_in = (struct of10_packet_in *)msg_in;
        eh = (struct ether_header *)(p_in->data);
        
#ifdef DEBUG
        printf("in_port: %hu, ", ntohs(p_in->in_port));
        printf("%02x:%02x:%02x:%02x:%02x:%02x ->", eh->ether_shost[0], eh->ether_shost[1],
                eh->ether_shost[2], eh->ether_shost[3], eh->ether_shost[4],
                eh->ether_shost[5]);
        printf(" %02x:%02x:%02x:%02x:%02x:%02x, ", eh->ether_dhost[0], eh->ether_dhost[1],
                eh->ether_dhost[2], eh->ether_dhost[3], eh->ether_dhost[4],
                eh->ether_dhost[5]);
        printf("type: 0x%hx ", ntohs(eh->ether_type));
        if (ntohs(eh->ether_type) == ETHERTYPE_ARP)
                printf("[ARP]");
        else if (ntohs(eh->ether_type) == ETHERTYPE_IP)
                printf("[IP4]");
        printf("\n");
        printf("packet_in reason: 0x%x\n", (unsigned int)p_in->reason);
#endif

        /* learn */
        learn(dp, p_in);

        /* check if it's an arp packet */
        if (ntohs(eh->ether_type) == ETHERTYPE_ARP)
                handle_arp(ctl, dp, p_in);
        /* check if it's an ip packet */
        else if (ntohs(eh->ether_type) == ETHERTYPE_IP)
                handle_ip(ctl, dp, p_in);
}

static void
handle_ofp_messages(struct of_controller *ctl, struct of_dataplane *dp, struct ofp_header *msg_in)
{
        printf("ofp_messages [dpid = 0x%016llx, type=0x%x].\n", DPID(dp), (unsigned int)msg_in->type);
        if (msg_in->type == OFPT10_PACKET_IN)
                handle_packet_in(ctl, dp, msg_in);
        else if (msg_in->type == OFPT10_STATS_REPLY)
                handle_stats(ctl, dp, msg_in);
}

static struct of10_packet_out *
arp_request(struct in_addr addr)
{
        struct ether_header             *eh;
        struct ether_arp                *arp;
        struct of10_packet_out          *msg;
        struct of10_action_output       *action;
        uint8_t                         *p;

        msg = (struct of10_packet_out *)
                malloc(sizeof(struct of10_packet_out) + sizeof(struct of10_action_output)
                + ETHER_HDR_LEN + sizeof(struct ether_arp));
        if (msg == NULL)
                return NULL;

        p = (uint8_t *)(msg) + sizeof(struct of10_packet_out) +
                sizeof(struct of10_action_output);
        eh = (struct ether_header *)p;
        p += ETHER_HDR_LEN;
        arp = (struct ether_arp *)p;

        /* ether_arp */
        arp->ea_hdr.ar_hrd = htons(ARPHRD_ETHER);
        arp->ea_hdr.ar_pro = htons(ETHERTYPE_IP);
        arp->ea_hdr.ar_op = htons(ARPOP_REQUEST);
        arp->ea_hdr.ar_hln = ETHER_ADDR_LEN;
        arp->ea_hdr.ar_pln = sizeof(in_addr_t);
        memset(arp->arp_tha, 0xff, ETHER_ADDR_LEN);
        memcpy(arp->arp_tpa, &addr.s_addr, sizeof(arp->arp_tpa));
        memcpy(arp->arp_sha, &vserver_mac, ETHER_ADDR_LEN);
        memcpy(arp->arp_spa, &vserver_ip.s_addr, sizeof(arp->arp_spa));
        /* ether_header */
        eh->ether_type = ntohs(ETHERTYPE_ARP);
        memset(eh->ether_dhost, 0xff, ETHER_ADDR_LEN);
        memcpy(eh->ether_shost, &vserver_mac, ETHER_ADDR_LEN);
        /* packet_out */        
        msg->hdr.version = OFP_VERSION_10;
        msg->hdr.type = OFPT10_PACKET_OUT;
        msg->hdr.length = htons(sizeof(struct of10_packet_out) + sizeof(struct of10_action_output) + ETHER_HDR_LEN + sizeof(struct ether_arp));
        msg->hdr.xid = 0;
        msg->buffer_id = htonl(-1);
        msg->in_port = 0xffff;  /* XXX: OFPP10_NONE */
        msg->actions_len = htons(sizeof(struct of10_action_output));
        action = (struct of10_action_output *)(msg->actions);
        action->type = htons(OFPAT10_OUTPUT);
        action->len = htons(sizeof(struct of10_action_output));
        action->port = htons(OFPP10_FLOOD);
        action->max_len = 0;
        return msg;
}

static void
vserver_stats_request(struct of_controller *ctl, struct of_dataplane *dp, uint16_t port)
{
        struct of10_stats_request       *msg;
        struct of10_port_stats_request  *body;

        msg = (struct of10_stats_request *)malloc(sizeof(struct of10_stats_request) + sizeof(struct of10_port_stats_request));
        if (msg == NULL)
                return;
        /* header */
        msg->hdr.version = OFP_VERSION_10;
        msg->hdr.type = OFPT10_STATS_REQUEST;
        msg->hdr.length = htons(sizeof(struct of10_stats_request) + sizeof(struct of10_port_stats_request));
        msg->hdr.xid = 0;
        /* type */
        msg->type = htons(OFPST10_PORT);
        /* flags */
        msg->flags = 0;
        /* body */
        body = (struct of10_port_stats_request *)(msg->body);
        /* port */
        body->port_no = htons(port);
        /* send message */
        ctl->send(ctl, dp, (struct ofp_header *)msg);
        free(msg);
}

static struct of_dataplane *
dpid2ptr(uint64_t dpid)
{
        struct dp_ptr           *dptr;

        SLIST_FOREACH(dptr, &dataplane_list, next)
                if ((dptr->dp) && (DPID(dptr->dp) == dpid))
                        return dptr->dp;
        return NULL;
}

static void
poll(struct of_controller *ctl)
{
        char                     *loadstr;
        int                       i, n, ret, local;
        struct host_info        **hosts;
        struct of_dataplane     *dp;
        struct server           *srv;


        /* read static topology information */
        ctx->topo(ctx, &hosts, &n);

        /* XXX: poll local servers only */
        SLIST_FOREACH(srv, &server_list, next) {
                if (!srv->ip_resolved)
                        continue;
                local = 0;
                for (i = 0 ; i < n ; i++) {
                        if (srv->ip_addr.s_addr == hosts[i]->ip_addr.s_addr) {
                                local = 1;
                                dp = dpid2ptr(hosts[i]->dpid);
                                if (!dp)
                                        continue;
#ifdef DEBUG
                                printf("[DEBUG] poll(): server %s at 0x%016llx.\n", inet_ntoa(srv->ip_addr), DPID(dp));
#endif
                                vserver_stats_request(ctl, dp, hosts[i]->port);
                                break;  /* XXX: support multi-connections */
                        }
                }
                /* check dht for remote servers */
                if (!local) {
                        ret = ctx->get(ctx, inet_ntoa(srv->ip_addr), &loadstr);
                        if (!ret) {
                                printf("[QUERY] get(%s) failed.\n", inet_ntoa(srv->ip_addr));
                        }
                        else {
                                printf("[QUERY] get(%s) = %s succeeded.\n", inet_ntoa(srv->ip_addr), loadstr);
                                sscanf(loadstr, "%llu", &srv->load);
                        }
                }
        }
}

static void
handle_connection_up(struct of_controller *ctl, struct of_dataplane *dp)
{
        struct ofp_header       *msg;
        struct server           *srv;
        struct dp_ptr           *dptr;
        uint64_t                 dpid;

        dpid = DPID(dp);
        printf("[INFO] connection_up [dpid = 0x%016llx].\n", dpid);

        if (auto_arp) {
                printf("[INFO] auto_arp is ON.\n");
                SLIST_FOREACH(srv, &server_list, next) {
                        printf("[INFO] learning about server %s.\n", inet_ntoa(srv->ip_addr));
                        msg = arp_request(srv->ip_addr);
                        ctl->send(ctl, dp, (struct ofp_header *)msg);
                        free(msg);
                }
        }
        dptr = (struct dp_ptr *)malloc(sizeof(struct dp_ptr));
        dptr->dp = dp;
        SLIST_INSERT_HEAD(&dataplane_list, dptr, next);
        printf("[INFO] switch [dpid = 0x%016llx] is ready.\n", dpid);
}

int
init(struct actl_ctx *c)
{
        ctx = c;
        printf("l3_lb initialized.\n");
        return 0;
}

int
main(int argc, char **argv)
{
        int                      i, ch, poll_period = DEFAULT_POLL_PERIOD;
        double                   tar_kpi;
        const char              *errstr;
        struct ether_addr       *ea;
        struct server           *srv;


#ifdef DEBUG
        for (i = 0 ; i < argc ; i++) {
                printf("arg[%d] = %s.\n", i, argv[i]);
        }
#endif

        SLIST_INIT(&dataplane_list);
        SLIST_INIT(&server_list);
        while ((ch = getopt_long(argc, argv, "i:m:s:a:p:k:t:h:", longopts, NULL)) != -1) {
                switch (ch) {
                case 'i':
                        if (!inet_aton(optarg, &vserver_ip))
                                errx(1, "invalid IP address");
                        break;
                case 'm':
                        if ((ea = ether_aton(optarg)) == NULL)
                                errx(1, "invalid MAC address");
                        vserver_mac = *ea;
                        break;
                case 's':
                        srv = (struct server *)malloc(sizeof(struct server));
                        if (!inet_aton(optarg, &srv->ip_addr))
                                errx(1, "invalid IP address");
                        srv->ip_resolved = 0;
                        srv->load = 0;
                        srv->load_bytes = 0;
                        srv->load_ts = time(NULL);
                        SLIST_INSERT_HEAD(&server_list, srv, next);
                        ++n_servers;
                        break;
                case 'a':
                        if (!strcmp(optarg, "yes"))
                                auto_arp = 1;
                        else if (!strcmp(optarg, "no"))
                                auto_arp = 0;
                        else
                                errx(1, "incorrect option for argument '-a'");
                        break;
                case 'p':
                        poll_period = strtonum(optarg, 1, 60, &errstr);
                        if (errstr != NULL)
                                errx(1, "invalid polling period %s: %s", errstr, optarg);
                        break;
                case 'k':
                        tar_kpi = (double)strtonum(optarg, 1, 100000, &errstr);
                        if (errstr)
                                errx(1, "invalid target KPI %s:%s", errstr, optarg);
                        ctx->adapt(ctx, tar_kpi);
                        break;
                case 't':
                        ttl_idle = (uint16_t)strtonum(optarg, 0, 1024, &errstr);
                        if (errstr)
                                errx(1, "flow idle ttl out-of-range %s:%s", errstr, optarg);
                        break;
                case 'h':
                        ttl_hard = (uint16_t)strtonum(optarg, 0, 1024, &errstr);
                        if (errstr)
                                errx(1, "flow hard ttl out-of-range %s:%s", errstr, optarg);
                        break;
                }
                /* TODO: default -> usage() */
        }
        argc -= optind;
        argv += optind;

        /* initialize the mactable */
        if (!hashtab_init(&mactable, 12, NULL))
                return 1;

        printf("Virtual Server IP: %s\n", inet_ntoa(vserver_ip));
        printf("Virtual Server MAC: %s\n", ether_ntoa(&vserver_mac));
        printf("Number of Physical Servers = %d.\n", n_servers);
        i = 0;
        SLIST_FOREACH(srv, &server_list, next)
                printf("Server%d IP: %s, MAC: %s\n", ++i, inet_ntoa(srv->ip_addr), ether_ntoa(&srv->mac_addr));
        printf("Target KPI = %lf.\n", tar_kpi);
        printf("TTL IDLE = %hu, HARD = %hu.\n", ttl_idle, ttl_hard);

        printf("L3 LB initialized.\n");

        /* polling timer */
        ctx->cntl->timer(ctx->cntl, poll_period * 1000, poll);

        return 0;
}

void
handler(struct of_event *ev)
{
#ifdef DEBUG
        printf("l3_lb_handler.\n");
#endif
        if (ev->type == OFEV_CONNECTION_UP)
                handle_connection_up(ctx->cntl, ev->dp);
        else if (ev->type == OFEV_PROTO_MESSAGE)
                handle_ofp_messages(ctx->cntl, ev->dp, ev->ofp_hdr);
}

double
kpi()
{
        double                   avg, stddev = 0;
        int                      n = 0;
        struct server           *srv;
        uint64_t                 load = 0;

#ifdef DEBUG
        printf("l3_lb_kpi.\n");
#endif
        if (!n_servers)
                return 0;

        /* mean */
        srv = SLIST_FIRST(&server_list);
        SLIST_FOREACH(srv, &server_list, next) {
                if (!srv->ip_resolved)
                        continue;
                load += srv->load;
                n++;
        }
        if (!n)
                return 0;
        avg = load / n;
        /* stddev */
        srv = SLIST_FIRST(&server_list);
        SLIST_FOREACH(srv, &server_list, next) {
                if (!srv->ip_resolved)
                        continue;
                stddev += (srv->load - avg) * (srv->load - avg);
        }
        stddev /= n;
        stddev = sqrt(stddev);
#ifdef DEBUG
        printf("l3_lb_kpi: KPI = %lf.\n", stddev);
#endif
        return stddev;
}