X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=blobdiff_plain;f=ofproto%2Fofproto.c;h=68b2493def2686c41dd661d8a8c9c430dd919808;hb=ca0f572cfe87f284018e14fa7f1de58fbaef4c87;hp=52e4fe3b622ad9f9ec435fceb1c1419f512f7444;hpb=abdfe47476cc1a192e329f428b3740a3fae8390d;p=openvswitch

diff --git a/ofproto/ofproto.c b/ofproto/ofproto.c
index 52e4fe3b..68b2493d 100644
--- a/ofproto/ofproto.c
+++ b/ofproto/ofproto.c
@@ -30,6 +30,8 @@
 #include "dpif.h"
 #include "dynamic-string.h"
 #include "fail-open.h"
+#include "hash.h"
+#include "hmap.h"
 #include "in-band.h"
 #include "mac-learning.h"
 #include "netdev.h"
@@ -46,11 +48,9 @@
 #include "pinsched.h"
 #include "pktbuf.h"
 #include "poll-loop.h"
-#include "port-array.h"
 #include "rconn.h"
 #include "shash.h"
 #include "status.h"
-#include "stp.h"
 #include "stream-ssl.h"
 #include "svec.h"
 #include "tag.h"
@@ -69,9 +69,12 @@ enum {
     TABLEID_CLASSIFIER = 1
 };
 
+
 struct ofport {
+    struct hmap_node hmap_node; /* In struct ofproto's "ports" hmap. */
     struct netdev *netdev;
     struct ofp_phy_port opp;    /* In host byte order. */
+    uint16_t odp_port;
 };
 
 static void ofport_free(struct ofport *);
@@ -86,7 +89,7 @@ static int xlate_actions(const union ofp_action *in, size_t n_in,
 struct rule {
     struct cls_rule cr;
 
-    uint64_t flow_cookie;       /* Controller-issued identifier. 
+    uint64_t flow_cookie;       /* Controller-issued identifier.
                                    (Kept in network-byte order.) */
     uint16_t idle_timeout;      /* In seconds from time of last use. */
     uint16_t hard_timeout;      /* In seconds from time of creation. */
@@ -170,19 +173,44 @@ static void send_flow_removed(struct ofproto *p, struct rule *rule,
 
 /* ofproto supports two kinds of OpenFlow connections:
  *
- *   - "Controller connections": Connections to ordinary OpenFlow controllers.
- *     ofproto maintains persistent connections to these controllers and by
- *     default sends them asynchronous messages such as packet-ins.
+ *   - "Primary" connections to ordinary OpenFlow controllers.  ofproto
+ *     maintains persistent connections to these controllers and by default
+ *     sends them asynchronous messages such as packet-ins.
  *
- *   - "Transient connections", e.g. from ovs-ofctl.  When these connections
+ *   - "Service" connections, e.g. from ovs-ofctl.  When these connections
  *     drop, it is the other side's responsibility to reconnect them if
  *     necessary.  ofproto does not send them asynchronous messages by default.
+ *
+ * Currently, active (tcp, ssl, unix) connections are always "primary"
+ * connections and passive (ptcp, pssl, punix) connections are always "service"
+ * connections.  There is no inherent reason for this, but it reflects the
+ * common case.
  */
 enum ofconn_type {
-    OFCONN_CONTROLLER,          /* An OpenFlow controller. */
-    OFCONN_TRANSIENT            /* A transient connection. */
+    OFCONN_PRIMARY,             /* An ordinary OpenFlow controller. */
+    OFCONN_SERVICE              /* A service connection, e.g. "ovs-ofctl". */
+};
+
+/* A listener for incoming OpenFlow "service" connections. */
+struct ofservice {
+    struct hmap_node node;      /* In struct ofproto's "services" hmap. */
+    struct pvconn *pvconn;      /* OpenFlow connection listener. */
+
+    /* These are not used by ofservice directly.  They are settings for
+     * accepted "struct ofconn"s from the pvconn. */
+    int probe_interval;         /* Max idle time before probing, in seconds. */
+    int rate_limit;             /* Max packet-in rate in packets per second. */
+    int burst_limit;            /* Limit on accumulating packet credits. */
 };
 
+static struct ofservice *ofservice_lookup(struct ofproto *,
+                                          const char *target);
+static int ofservice_create(struct ofproto *,
+                            const struct ofproto_controller *);
+static void ofservice_reconfigure(struct ofservice *,
+                                  const struct ofproto_controller *);
+static void ofservice_destroy(struct ofproto *, struct ofservice *);
+
 /* An OpenFlow connection. */
 struct ofconn {
     struct ofproto *ofproto;    /* The ofproto that owns this connection. */
@@ -202,7 +230,7 @@ struct ofconn {
 #define OFCONN_REPLY_MAX 100
     struct rconn_packet_counter *reply_counter;
 
-    /* type == OFCONN_CONTROLLER only. */
+    /* type == OFCONN_PRIMARY only. */
     enum nx_role role;           /* Role. */
     struct hmap_node hmap_node;  /* In struct ofproto's "controllers" map. */
     struct discovery *discovery; /* Controller discovery object, if enabled. */
@@ -227,6 +255,7 @@ static void ofconn_run(struct ofconn *, struct ofproto *);
 static void ofconn_wait(struct ofconn *);
 static bool ofconn_receives_async_msgs(const struct ofconn *);
 static char *ofconn_make_name(const struct ofproto *, const char *target);
+static void ofconn_set_rate_limit(struct ofconn *, int rate, int burst);
 
 static void queue_tx(struct ofpbuf *msg, const struct ofconn *ofconn,
                      struct rconn_packet_counter *counter);
@@ -247,8 +276,7 @@ struct ofproto {
     /* Datapath. */
     struct dpif *dpif;
     struct netdev_monitor *netdev_monitor;
-    struct port_array ports;    /* Index is ODP port nr; ofport->opp.port_no is
-                                 * OFP port nr. */
+    struct hmap ports;          /* Contains "struct ofport"s. */
     struct shash port_by_name;
     uint32_t max_ports;
 
@@ -275,8 +303,9 @@ struct ofproto {
     struct hmap controllers;   /* Controller "struct ofconn"s. */
     struct list all_conns;     /* Contains "struct ofconn"s. */
     enum ofproto_fail_mode fail_mode;
-    struct pvconn **listeners;
-    size_t n_listeners;
+
+    /* OpenFlow listeners. */
+    struct hmap services;       /* Contains "struct ofservice"s. */
     struct pvconn **snoops;
     size_t n_snoops;
 
@@ -310,6 +339,7 @@ static void handle_openflow(struct ofconn *, struct ofproto *,
 
 static void refresh_port_groups(struct ofproto *);
 
+static struct ofport *get_port(const struct ofproto *, uint16_t odp_port);
 static void update_port(struct ofproto *, const char *devname);
 static int init_ports(struct ofproto *);
 static void reinit_ports(struct ofproto *);
@@ -362,7 +392,7 @@ ofproto_create(const char *datapath, const char *datapath_type,
     /* Initialize datapath. */
     p->dpif = dpif;
     p->netdev_monitor = netdev_monitor_create();
-    port_array_init(&p->ports);
+    hmap_init(&p->ports);
     shash_init(&p->port_by_name);
     p->max_ports = stats.max_ports;
 
@@ -382,8 +412,7 @@ ofproto_create(const char *datapath, const char *datapath_type,
     /* Initialize OpenFlow connections. */
     list_init(&p->all_conns);
     hmap_init(&p->controllers);
-    p->listeners = NULL;
-    p->n_listeners = 0;
+    hmap_init(&p->services);
     p->snoops = NULL;
     p->n_snoops = 0;
 
@@ -452,7 +481,7 @@ add_controller(struct ofproto *ofproto, const struct ofproto_controller *c)
         discovery = NULL;
     }
 
-    ofconn = ofconn_create(ofproto, rconn_create(5, 8), OFCONN_CONTROLLER);
+    ofconn = ofconn_create(ofproto, rconn_create(5, 8), OFCONN_PRIMARY);
     ofconn->pktbuf = pktbuf_create();
     ofconn->miss_send_len = OFP_DEFAULT_MISS_SEND_LEN;
     if (discovery) {
@@ -473,9 +502,7 @@ add_controller(struct ofproto *ofproto, const struct ofproto_controller *c)
 static void
 update_controller(struct ofconn *ofconn, const struct ofproto_controller *c)
 {
-    struct ofproto *ofproto = ofconn->ofproto;
     int probe_interval;
-    int i;
 
     ofconn->band = (is_in_band_controller(c)
                     ? OFPROTO_IN_BAND : OFPROTO_OUT_OF_BAND);
@@ -491,21 +518,7 @@ update_controller(struct ofconn *ofconn, const struct ofproto_controller *c)
         discovery_set_accept_controller_re(ofconn->discovery, c->accept_re);
     }
 
-    for (i = 0; i < N_SCHEDULERS; i++) {
-        struct pinsched **s = &ofconn->schedulers[i];
-
-        if (c->rate_limit > 0) {
-            if (!*s) {
-                *s = pinsched_create(c->rate_limit, c->burst_limit,
-                                     ofproto->switch_status);
-            } else {
-                pinsched_set_limits(*s, c->rate_limit, c->burst_limit);
-            }
-        } else {
-            pinsched_destroy(*s);
-            *s = NULL;
-        }
-    }
+    ofconn_set_rate_limit(ofconn, c->rate_limit, c->burst_limit);
 }
 
 static const char *
@@ -621,22 +634,38 @@ ofproto_set_controllers(struct ofproto *p,
                         size_t n_controllers)
 {
     struct shash new_controllers;
-    struct ofconn *ofconn, *next;
+    struct ofconn *ofconn, *next_ofconn;
+    struct ofservice *ofservice, *next_ofservice;
     bool ss_exists;
     size_t i;
 
+    /* Create newly configured controllers and services.
+     * Create a name to ofproto_controller mapping in 'new_controllers'. */
     shash_init(&new_controllers);
     for (i = 0; i < n_controllers; i++) {
         const struct ofproto_controller *c = &controllers[i];
 
-        shash_add_once(&new_controllers, c->target, &controllers[i]);
-        if (!find_controller_by_target(p, c->target)) {
-            add_controller(p, c);
+        if (!vconn_verify_name(c->target) || !strcmp(c->target, "discover")) {
+            if (!find_controller_by_target(p, c->target)) {
+                add_controller(p, c);
+            }
+        } else if (!pvconn_verify_name(c->target)) {
+            if (!ofservice_lookup(p, c->target) && ofservice_create(p, c)) {
+                continue;
+            }
+        } else {
+            VLOG_WARN_RL(&rl, "%s: unsupported controller \"%s\"",
+                         dpif_name(p->dpif), c->target);
+            continue;
         }
+
+        shash_add_once(&new_controllers, c->target, &controllers[i]);
     }
 
+    /* Delete controllers that are no longer configured.
+     * Update configuration of all now-existing controllers. */
     ss_exists = false;
-    HMAP_FOR_EACH_SAFE (ofconn, next, struct ofconn, hmap_node,
+    HMAP_FOR_EACH_SAFE (ofconn, next_ofconn, struct ofconn, hmap_node,
                         &p->controllers) {
         struct ofproto_controller *c;
 
@@ -650,10 +679,25 @@ ofproto_set_controllers(struct ofproto *p,
             }
         }
     }
+
+    /* Delete services that are no longer configured.
+     * Update configuration of all now-existing services. */
+    HMAP_FOR_EACH_SAFE (ofservice, next_ofservice, struct ofservice, node,
+                        &p->services) {
+        struct ofproto_controller *c;
+
+        c = shash_find_data(&new_controllers,
+                            pvconn_get_name(ofservice->pvconn));
+        if (!c) {
+            ofservice_destroy(p, ofservice);
+        } else {
+            ofservice_reconfigure(ofservice, c);
+        }
+    }
+
     shash_destroy(&new_controllers);
 
     update_in_band_remotes(p);
-
     update_fail_open(p);
 
     if (!hmap_is_empty(&p->controllers) && !ss_exists) {
@@ -813,12 +857,6 @@ set_pvconns(struct pvconn ***pvconnsp, size_t *n_pvconnsp,
     return retval;
 }
 
-int
-ofproto_set_listeners(struct ofproto *ofproto, const struct svec *listeners)
-{
-    return set_pvconns(&ofproto->listeners, &ofproto->n_listeners, listeners);
-}
-
 int
 ofproto_set_snoops(struct ofproto *ofproto, const struct svec *snoops)
 {
@@ -849,12 +887,11 @@ ofproto_set_sflow(struct ofproto *ofproto,
     if (oso) {
         if (!os) {
             struct ofport *ofport;
-            unsigned int odp_port;
 
             os = ofproto->sflow = ofproto_sflow_create(ofproto->dpif);
             refresh_port_groups(ofproto);
-            PORT_ARRAY_FOR_EACH (ofport, &ofproto->ports, odp_port) {
-                ofproto_sflow_add_port(os, odp_port,
+            HMAP_FOR_EACH (ofport, struct ofport, hmap_node, &ofproto->ports) {
+                ofproto_sflow_add_port(os, ofport->odp_port,
                                        netdev_get_name(ofport->netdev));
             }
         }
@@ -865,18 +902,6 @@ ofproto_set_sflow(struct ofproto *ofproto,
     }
 }
 
-int
-ofproto_set_stp(struct ofproto *ofproto OVS_UNUSED, bool enable_stp)
-{
-    /* XXX */
-    if (enable_stp) {
-        VLOG_WARN("STP is not yet implemented");
-        return EINVAL;
-    } else {
-        return 0;
-    }
-}
-
 uint64_t
 ofproto_get_datapath_id(const struct ofproto *ofproto)
 {
@@ -884,7 +909,7 @@ ofproto_get_datapath_id(const struct ofproto *ofproto)
 }
 
 bool
-ofproto_has_controller(const struct ofproto *ofproto)
+ofproto_has_primary_controller(const struct ofproto *ofproto)
 {
     return !hmap_is_empty(&ofproto->controllers);
 }
@@ -895,16 +920,6 @@ ofproto_get_fail_mode(const struct ofproto *p)
     return p->fail_mode;
 }
 
-void
-ofproto_get_listeners(const struct ofproto *ofproto, struct svec *listeners)
-{
-    size_t i;
-
-    for (i = 0; i < ofproto->n_listeners; i++) {
-        svec_add(listeners, pvconn_get_name(ofproto->listeners[i]));
-    }
-}
-
 void
 ofproto_get_snoops(const struct ofproto *ofproto, struct svec *snoops)
 {
@@ -918,9 +933,9 @@ ofproto_get_snoops(const struct ofproto *ofproto, struct svec *snoops)
 void
 ofproto_destroy(struct ofproto *p)
 {
+    struct ofservice *ofservice, *next_ofservice;
     struct ofconn *ofconn, *next_ofconn;
-    struct ofport *ofport;
-    unsigned int port_no;
+    struct ofport *ofport, *next_ofport;
     size_t i;
 
     if (!p) {
@@ -946,7 +961,9 @@ ofproto_destroy(struct ofproto *p)
 
     dpif_close(p->dpif);
     netdev_monitor_destroy(p->netdev_monitor);
-    PORT_ARRAY_FOR_EACH (ofport, &p->ports, port_no) {
+    HMAP_FOR_EACH_SAFE (ofport, next_ofport, struct ofport, hmap_node,
+                        &p->ports) {
+        hmap_remove(&p->ports, &ofport->hmap_node);
         ofport_free(ofport);
     }
     shash_destroy(&p->port_by_name);
@@ -955,10 +972,11 @@ ofproto_destroy(struct ofproto *p)
     netflow_destroy(p->netflow);
     ofproto_sflow_destroy(p->sflow);
 
-    for (i = 0; i < p->n_listeners; i++) {
-        pvconn_close(p->listeners[i]);
+    HMAP_FOR_EACH_SAFE (ofservice, next_ofservice, struct ofservice, node,
+                        &p->services) {
+        ofservice_destroy(p, ofservice);
     }
-    free(p->listeners);
+    hmap_destroy(&p->services);
 
     for (i = 0; i < p->n_snoops; i++) {
         pvconn_close(p->snoops[i]);
@@ -973,7 +991,7 @@ ofproto_destroy(struct ofproto *p)
     free(p->serial_desc);
     free(p->dp_desc);
 
-    port_array_destroy(&p->ports);
+    hmap_destroy(&p->ports);
 
     free(p);
 }
@@ -1028,7 +1046,7 @@ add_snooper(struct ofproto *ofproto, struct vconn *vconn)
     /* Pick a controller for monitoring. */
     best = NULL;
     LIST_FOR_EACH (ofconn, struct ofconn, node, &ofproto->all_conns) {
-        if (ofconn->type == OFCONN_CONTROLLER
+        if (ofconn->type == OFCONN_PRIMARY
             && (!best || snoop_preference(ofconn) > snoop_preference(best))) {
             best = ofconn;
         }
@@ -1046,6 +1064,7 @@ int
 ofproto_run1(struct ofproto *p)
 {
     struct ofconn *ofconn, *next_ofconn;
+    struct ofservice *ofservice;
     char *devname;
     int error;
     int i;
@@ -1056,7 +1075,6 @@ ofproto_run1(struct ofproto *p)
 
     for (i = 0; i < 50; i++) {
         struct ofpbuf *buf;
-        int error;
 
         error = dpif_recv(p->dpif, &buf);
         if (error) {
@@ -1101,21 +1119,23 @@ ofproto_run1(struct ofproto *p)
         fail_open_run(p->fail_open);
     }
 
-    for (i = 0; i < p->n_listeners; i++) {
+    HMAP_FOR_EACH (ofservice, struct ofservice, node, &p->services) {
         struct vconn *vconn;
         int retval;
 
-        retval = pvconn_accept(p->listeners[i], OFP_VERSION, &vconn);
+        retval = pvconn_accept(ofservice->pvconn, OFP_VERSION, &vconn);
         if (!retval) {
             struct rconn *rconn;
             char *name;
 
-            rconn = rconn_create(60, 0);
+            rconn = rconn_create(ofservice->probe_interval, 0);
             name = ofconn_make_name(p, vconn_get_name(vconn));
             rconn_connect_unreliably(rconn, vconn, name);
             free(name);
 
-            ofconn_create(p, rconn, OFCONN_TRANSIENT);
+            ofconn = ofconn_create(p, rconn, OFCONN_SERVICE);
+            ofconn_set_rate_limit(ofconn, ofservice->rate_limit,
+                                  ofservice->burst_limit);
         } else if (retval != EAGAIN) {
             VLOG_WARN_RL(&rl, "accept failed (%s)", strerror(retval));
         }
@@ -1188,6 +1208,7 @@ ofproto_run2(struct ofproto *p, bool revalidate_all)
 void
 ofproto_wait(struct ofproto *p)
 {
+    struct ofservice *ofservice;
     struct ofconn *ofconn;
     size_t i;
 
@@ -1217,8 +1238,8 @@ ofproto_wait(struct ofproto *p)
     } else if (p->next_expiration != LLONG_MAX) {
         poll_timer_wait_until(p->next_expiration);
     }
-    for (i = 0; i < p->n_listeners; i++) {
-        pvconn_wait(p->listeners[i]);
+    HMAP_FOR_EACH (ofservice, struct ofservice, node, &p->services) {
+        pvconn_wait(ofservice->pvconn);
     }
     for (i = 0; i < p->n_snoops; i++) {
         pvconn_wait(p->snoops[i]);
@@ -1272,7 +1293,7 @@ ofproto_add_flow(struct ofproto *p,
 {
     struct rule *rule;
     rule = rule_create(p, NULL, actions, n_actions,
-                       idle_timeout >= 0 ? idle_timeout : 5 /* XXX */, 
+                       idle_timeout >= 0 ? idle_timeout : 5 /* XXX */,
                        0, 0, false);
     cls_rule_from_flow(flow, wildcards, priority, &rule->cr);
     rule_insert(p, rule, NULL, 0);
@@ -1326,13 +1347,12 @@ reinit_ports(struct ofproto *p)
 {
     struct svec devnames;
     struct ofport *ofport;
-    unsigned int port_no;
     struct odp_port *odp_ports;
     size_t n_odp_ports;
     size_t i;
 
     svec_init(&devnames);
-    PORT_ARRAY_FOR_EACH (ofport, &p->ports, port_no) {
+    HMAP_FOR_EACH (ofport, struct ofport, hmap_node, &p->ports) {
         svec_add (&devnames, (char *) ofport->opp.name);
     }
     dpif_port_list(p->dpif, &odp_ports, &n_odp_ports);
@@ -1354,15 +1374,14 @@ refresh_port_group(struct ofproto *p, unsigned int group)
     uint16_t *ports;
     size_t n_ports;
     struct ofport *port;
-    unsigned int port_no;
 
     assert(group == DP_GROUP_ALL || group == DP_GROUP_FLOOD);
 
-    ports = xmalloc(port_array_count(&p->ports) * sizeof *ports);
+    ports = xmalloc(hmap_count(&p->ports) * sizeof *ports);
     n_ports = 0;
-    PORT_ARRAY_FOR_EACH (port, &p->ports, port_no) {
+    HMAP_FOR_EACH (port, struct ofport, hmap_node, &p->ports) {
         if (group == DP_GROUP_ALL || !(port->opp.config & OFPPC_NO_FLOOD)) {
-            ports[n_ports++] = port_no;
+            ports[n_ports++] = port->odp_port;
         }
     }
     dpif_port_group_set(p->dpif, group, ports, n_ports);
@@ -1406,6 +1425,7 @@ make_ofport(const struct odp_port *odp_port)
 
     ofport = xmalloc(sizeof *ofport);
     ofport->netdev = netdev;
+    ofport->odp_port = odp_port->port;
     ofport->opp.port_no = odp_port_to_ofp_port(odp_port->port);
     netdev_get_etheraddr(netdev, ofport->opp.hw_addr);
     memcpy(ofport->opp.name, odp_port->devname,
@@ -1427,7 +1447,7 @@ make_ofport(const struct odp_port *odp_port)
 static bool
 ofport_conflicts(const struct ofproto *p, const struct odp_port *odp_port)
 {
-    if (port_array_get(&p->ports, odp_port->port)) {
+    if (get_port(p, odp_port->port)) {
         VLOG_WARN_RL(&rl, "ignoring duplicate port %"PRIu16" in datapath",
                      odp_port->port);
         return true;
@@ -1486,28 +1506,25 @@ send_port_status(struct ofproto *p, const struct ofport *ofport,
 static void
 ofport_install(struct ofproto *p, struct ofport *ofport)
 {
-    uint16_t odp_port = ofp_port_to_odp_port(ofport->opp.port_no);
     const char *netdev_name = (const char *) ofport->opp.name;
 
     netdev_monitor_add(p->netdev_monitor, ofport->netdev);
-    port_array_set(&p->ports, odp_port, ofport);
+    hmap_insert(&p->ports, &ofport->hmap_node, hash_int(ofport->odp_port, 0));
     shash_add(&p->port_by_name, netdev_name, ofport);
     if (p->sflow) {
-        ofproto_sflow_add_port(p->sflow, odp_port, netdev_name);
+        ofproto_sflow_add_port(p->sflow, ofport->odp_port, netdev_name);
     }
 }
 
 static void
 ofport_remove(struct ofproto *p, struct ofport *ofport)
 {
-    uint16_t odp_port = ofp_port_to_odp_port(ofport->opp.port_no);
-
     netdev_monitor_remove(p->netdev_monitor, ofport->netdev);
-    port_array_delete(&p->ports, odp_port);
+    hmap_remove(&p->ports, &ofport->hmap_node);
     shash_delete(&p->port_by_name,
                  shash_find(&p->port_by_name, (char *) ofport->opp.name));
     if (p->sflow) {
-        ofproto_sflow_del_port(p->sflow, odp_port);
+        ofproto_sflow_del_port(p->sflow, ofport->odp_port);
     }
 }
 
@@ -1520,6 +1537,20 @@ ofport_free(struct ofport *ofport)
     }
 }
 
+static struct ofport *
+get_port(const struct ofproto *ofproto, uint16_t odp_port)
+{
+    struct ofport *port;
+
+    HMAP_FOR_EACH_IN_BUCKET (port, struct ofport, hmap_node,
+                             hash_int(odp_port, 0), &ofproto->ports) {
+        if (port->odp_port == odp_port) {
+            return port;
+        }
+    }
+    return NULL;
+}
+
 static void
 update_port(struct ofproto *p, const char *devname)
 {
@@ -1547,7 +1578,7 @@ update_port(struct ofproto *p, const char *devname)
              * reliably but more portably by comparing the old port's MAC
              * against the new port's MAC.  However, this code isn't that smart
              * and always sends an OFPPR_MODIFY (XXX). */
-            old_ofport = port_array_get(&p->ports, odp_port.port);
+            old_ofport = get_port(p, odp_port.port);
         }
     } else if (error != ENOENT && error != ENODEV) {
         VLOG_WARN_RL(&rl, "dpif_port_query_by_name returned unexpected error "
@@ -1638,7 +1669,7 @@ ofconn_create(struct ofproto *p, struct rconn *rconn, enum ofconn_type type)
 static void
 ofconn_destroy(struct ofconn *ofconn)
 {
-    if (ofconn->type == OFCONN_CONTROLLER) {
+    if (ofconn->type == OFCONN_PRIMARY) {
         hmap_remove(&ofconn->ofproto->controllers, &ofconn->hmap_node);
     }
     discovery_destroy(ofconn->discovery);
@@ -1724,14 +1755,13 @@ ofconn_wait(struct ofconn *ofconn)
 static bool
 ofconn_receives_async_msgs(const struct ofconn *ofconn)
 {
-    if (ofconn->type == OFCONN_CONTROLLER) {
-        /* Ordinary controllers always get asynchronous messages unless they
+    if (ofconn->type == OFCONN_PRIMARY) {
+        /* Primary controllers always get asynchronous messages unless they
          * have configured themselves as "slaves".  */
         return ofconn->role != NX_ROLE_SLAVE;
     } else {
-        /* Transient connections don't get asynchronous messages unless they
-         * have explicitly asked for them by setting a nonzero miss send
-         * length. */
+        /* Service connections don't get asynchronous messages unless they have
+         * explicitly asked for them by setting a nonzero miss send length. */
         return ofconn->miss_send_len > 0;
     }
 }
@@ -1747,6 +1777,85 @@ ofconn_make_name(const struct ofproto *ofproto, const char *target)
 {
     return xasprintf("%s<->%s", dpif_base_name(ofproto->dpif), target);
 }
+
+static void
+ofconn_set_rate_limit(struct ofconn *ofconn, int rate, int burst)
+{
+    int i;
+
+    for (i = 0; i < N_SCHEDULERS; i++) {
+        struct pinsched **s = &ofconn->schedulers[i];
+
+        if (rate > 0) {
+            if (!*s) {
+                *s = pinsched_create(rate, burst,
+                                     ofconn->ofproto->switch_status);
+            } else {
+                pinsched_set_limits(*s, rate, burst);
+            }
+        } else {
+            pinsched_destroy(*s);
+            *s = NULL;
+        }
+    }
+}
+
+static void
+ofservice_reconfigure(struct ofservice *ofservice,
+                      const struct ofproto_controller *c)
+{
+    ofservice->probe_interval = c->probe_interval;
+    ofservice->rate_limit = c->rate_limit;
+    ofservice->burst_limit = c->burst_limit;
+}
+
+/* Creates a new ofservice in 'ofproto'.  Returns 0 if successful, otherwise a
+ * positive errno value. */
+static int
+ofservice_create(struct ofproto *ofproto, const struct ofproto_controller *c)
+{
+    struct ofservice *ofservice;
+    struct pvconn *pvconn;
+    int error;
+
+    error = pvconn_open(c->target, &pvconn);
+    if (error) {
+        return error;
+    }
+
+    ofservice = xzalloc(sizeof *ofservice);
+    hmap_insert(&ofproto->services, &ofservice->node,
+                hash_string(c->target, 0));
+    ofservice->pvconn = pvconn;
+
+    ofservice_reconfigure(ofservice, c);
+
+    return 0;
+}
+
+static void
+ofservice_destroy(struct ofproto *ofproto, struct ofservice *ofservice)
+{
+    hmap_remove(&ofproto->services, &ofservice->node);
+    pvconn_close(ofservice->pvconn);
+    free(ofservice);
+}
+
+/* Finds and returns the ofservice within 'ofproto' that has the given
+ * 'target', or a null pointer if none exists. */
+static struct ofservice *
+ofservice_lookup(struct ofproto *ofproto, const char *target)
+{
+    struct ofservice *ofservice;
+
+    HMAP_FOR_EACH_WITH_HASH (ofservice, struct ofservice, node,
+                             hash_string(target, 0), &ofproto->services) {
+        if (!strcmp(pvconn_get_name(ofservice->pvconn), target)) {
+            return ofservice;
+        }
+    }
+    return NULL;
+}
 
 /* Caller is responsible for initializing the 'cr' member of the returned
  * rule. */
@@ -1830,8 +1939,44 @@ rule_has_out_port(const struct rule *rule, uint16_t out_port)
     return false;
 }
 
+/* Executes, within 'ofproto', the 'n_actions' actions in 'actions' on
+ * 'packet', which arrived on 'in_port'.
+ *
+ * Takes ownership of 'packet'. */
+static bool
+execute_odp_actions(struct ofproto *ofproto, uint16_t in_port,
+                    const union odp_action *actions, size_t n_actions,
+                    struct ofpbuf *packet)
+{
+    if (n_actions == 1 && actions[0].type == ODPAT_CONTROLLER) {
+        /* As an optimization, avoid a round-trip from userspace to kernel to
+         * userspace.  This also avoids possibly filling up kernel packet
+         * buffers along the way. */
+        struct odp_msg *msg;
+
+        msg = ofpbuf_push_uninit(packet, sizeof *msg);
+        msg->type = _ODPL_ACTION_NR;
+        msg->length = sizeof(struct odp_msg) + packet->size;
+        msg->port = in_port;
+        msg->reserved = 0;
+        msg->arg = actions[0].controller.arg;
+
+        send_packet_in(ofproto, packet);
+
+        return true;
+    } else {
+        int error;
+
+        error = dpif_execute(ofproto->dpif, in_port,
+                             actions, n_actions, packet);
+        ofpbuf_delete(packet);
+        return !error;
+    }
+}
+
 /* Executes the actions indicated by 'rule' on 'packet', which is in flow
- * 'flow' and is considered to have arrived on ODP port 'in_port'.
+ * 'flow' and is considered to have arrived on ODP port 'in_port'.  'packet'
+ * must have at least sizeof(struct ofp_packet_in) bytes of headroom.
  *
  * The flow that 'packet' actually contains does not need to actually match
  * 'rule'; the actions in 'rule' will be applied to it either way.  Likewise,
@@ -1843,15 +1988,20 @@ rule_has_out_port(const struct rule *rule, uint16_t out_port)
  * 'packet' using rule_make_actions().  If 'rule' is a wildcard rule, or if
  * 'rule' is an exact-match rule but 'flow' is not the rule's flow, then this
  * function will compose a set of ODP actions based on 'rule''s OpenFlow
- * actions and apply them to 'packet'. */
+ * actions and apply them to 'packet'.
+ *
+ * Takes ownership of 'packet'. */
 static void
 rule_execute(struct ofproto *ofproto, struct rule *rule,
              struct ofpbuf *packet, const flow_t *flow)
 {
     const union odp_action *actions;
+    struct odp_flow_stats stats;
     size_t n_actions;
     struct odp_actions a;
 
+    assert(ofpbuf_headroom(packet) >= sizeof(struct ofp_packet_in));
+
     /* Grab or compose the ODP actions.
      *
      * The special case for an exact-match 'rule' where 'flow' is not the
@@ -1862,6 +2012,7 @@ rule_execute(struct ofproto *ofproto, struct rule *rule,
         struct rule *super = rule->super ? rule->super : rule;
         if (xlate_actions(super->actions, super->n_actions, flow, ofproto,
                           packet, &a, NULL, 0, NULL)) {
+            ofpbuf_delete(packet);
             return;
         }
         actions = a.actions;
@@ -1872,16 +2023,21 @@ rule_execute(struct ofproto *ofproto, struct rule *rule,
     }
 
     /* Execute the ODP actions. */
-    if (!dpif_execute(ofproto->dpif, flow->in_port,
-                      actions, n_actions, packet)) {
-        struct odp_flow_stats stats;
-        flow_extract_stats(flow, packet, &stats);
+    flow_extract_stats(flow, packet, &stats);
+    if (execute_odp_actions(ofproto, flow->in_port,
+                            actions, n_actions, packet)) {
         update_stats(ofproto, rule, &stats);
         rule->used = time_msec();
         netflow_flow_update_time(ofproto->netflow, &rule->nf_flow, rule->used);
     }
 }
 
+/* Inserts 'rule' into 'p''s flow table.
+ *
+ * If 'packet' is nonnull, takes ownership of 'packet', executes 'rule''s
+ * actions on it and credits the statistics for sending the packet to 'rule'.
+ * 'packet' must have at least sizeof(struct ofp_packet_in) bytes of
+ * headroom. */
 static void
 rule_insert(struct ofproto *p, struct rule *rule, struct ofpbuf *packet,
             uint16_t in_port)
@@ -2062,7 +2218,7 @@ rule_account(struct ofproto *ofproto, struct rule *rule, uint64_t extra_bytes)
         && total_bytes > rule->accounted_bytes)
     {
         ofproto->ofhooks->account_flow_cb(
-            &rule->cr.flow, rule->odp_actions, rule->n_odp_actions,
+            &rule->cr.flow, rule->tags, rule->odp_actions, rule->n_odp_actions,
             total_bytes - rule->accounted_bytes, ofproto->aux);
         rule->accounted_bytes = total_bytes;
     }
@@ -2197,7 +2353,6 @@ handle_features_request(struct ofproto *p, struct ofconn *ofconn,
 {
     struct ofp_switch_features *osf;
     struct ofpbuf *buf;
-    unsigned int port_no;
     struct ofport *port;
 
     osf = make_openflow_xid(sizeof *osf, OFPT_FEATURES_REPLY, oh->xid, &buf);
@@ -2219,7 +2374,7 @@ handle_features_request(struct ofproto *p, struct ofconn *ofconn,
                          (1u << OFPAT_SET_TP_DST) |
                          (1u << OFPAT_ENQUEUE));
 
-    PORT_ARRAY_FOR_EACH (port, &p->ports, port_no) {
+    HMAP_FOR_EACH (port, struct ofport, hmap_node, &p->ports) {
         hton_ofp_phy_port(ofpbuf_put(buf, &port->opp, sizeof port->opp));
     }
 
@@ -2262,7 +2417,7 @@ handle_set_config(struct ofproto *p, struct ofconn *ofconn,
     }
     flags = ntohs(osc->flags);
 
-    if (ofconn->type == OFCONN_CONTROLLER && ofconn->role != NX_ROLE_SLAVE) {
+    if (ofconn->type == OFCONN_PRIMARY && ofconn->role != NX_ROLE_SLAVE) {
         switch (flags & OFPC_FRAG_MASK) {
         case OFPC_FRAG_NORMAL:
             dpif_set_drop_frags(p->dpif, false);
@@ -2317,13 +2472,17 @@ struct action_xlate_ctx {
     uint16_t nf_output_iface;   /* Output interface index for NetFlow. */
 };
 
+/* Maximum depth of flow table recursion (due to NXAST_RESUBMIT actions) in a
+ * flow translation. */
+#define MAX_RESUBMIT_RECURSION 8
+
 static void do_xlate_actions(const union ofp_action *in, size_t n_in,
                              struct action_xlate_ctx *ctx);
 
 static void
 add_output_action(struct action_xlate_ctx *ctx, uint16_t port)
 {
-    const struct ofport *ofport = port_array_get(&ctx->ofproto->ports, port);
+    const struct ofport *ofport = get_port(ctx->ofproto, port);
 
     if (ofport) {
         if (ofport->opp.config & OFPPC_NO_FWD) {
@@ -2364,7 +2523,7 @@ lookup_valid_rule(struct ofproto *ofproto, const flow_t *flow)
 static void
 xlate_table_action(struct action_xlate_ctx *ctx, uint16_t in_port)
 {
-    if (!ctx->recurse) {
+    if (ctx->recurse < MAX_RESUBMIT_RECURSION) {
         uint16_t old_in_port;
         struct rule *rule;
 
@@ -2385,6 +2544,11 @@ xlate_table_action(struct action_xlate_ctx *ctx, uint16_t in_port)
             do_xlate_actions(rule->actions, rule->n_actions, ctx);
             ctx->recurse--;
         }
+    } else {
+        struct vlog_rate_limit recurse_rl = VLOG_RATE_LIMIT_INIT(1, 1);
+
+        VLOG_ERR_RL(&recurse_rl, "NXAST_RESUBMIT recursed over %d times",
+                    MAX_RESUBMIT_RECURSION);
     }
 }
 
@@ -2525,6 +2689,12 @@ xlate_nicira_action(struct action_xlate_ctx *ctx,
         ctx->flow.tun_id = oa->tunnel.tun_id = nast->tun_id;
         break;
 
+    case NXAST_DROP_SPOOFED_ARP:
+        if (ctx->flow.dl_type == htons(ETH_TYPE_ARP)) {
+            odp_actions_add(ctx->out, ODPAT_DROP_SPOOFED_ARP);
+        }
+        break;
+
     /* If you add a new action here that modifies flow data, don't forget to
      * update the flow key in ctx->flow at the same time. */
 
@@ -2542,9 +2712,9 @@ do_xlate_actions(const union ofp_action *in, size_t n_in,
     const union ofp_action *ia;
     const struct ofport *port;
 
-    port = port_array_get(&ctx->ofproto->ports, ctx->flow.in_port);
+    port = get_port(ctx->ofproto, ctx->flow.in_port);
     if (port && port->opp.config & (OFPPC_NO_RECV | OFPPC_NO_RECV_STP) &&
-        port->opp.config & (eth_addr_equals(ctx->flow.dl_dst, stp_eth_addr)
+        port->opp.config & (eth_addr_equals(ctx->flow.dl_dst, eth_addr_stp)
                             ? OFPPC_NO_RECV_STP : OFPPC_NO_RECV)) {
         /* Drop this flow. */
         return;
@@ -2666,6 +2836,7 @@ xlate_actions(const union ofp_action *in, size_t n_in,
         *nf_output_iface = ctx.nf_output_iface;
     }
     if (odp_actions_overflow(out)) {
+        COVERAGE_INC(odp_overflow);
         odp_actions_init(out);
         return ofp_mkerr(OFPET_BAD_ACTION, OFPBAC_TOO_MANY);
     }
@@ -2680,7 +2851,7 @@ xlate_actions(const union ofp_action *in, size_t n_in,
 static int
 reject_slave_controller(struct ofconn *ofconn, const struct ofp_header *oh)
 {
-    if (ofconn->type == OFCONN_CONTROLLER && ofconn->role == NX_ROLE_SLAVE) {
+    if (ofconn->type == OFCONN_PRIMARY && ofconn->role == NX_ROLE_SLAVE) {
         static struct vlog_rate_limit perm_rl = VLOG_RATE_LIMIT_INIT(1, 5);
         char *type_name;
 
@@ -2790,8 +2961,7 @@ handle_port_mod(struct ofproto *p, struct ofconn *ofconn,
     }
     opm = (struct ofp_port_mod *) oh;
 
-    port = port_array_get(&p->ports,
-                          ofp_port_to_odp_port(ntohs(opm->port_no)));
+    port = get_port(p, ofp_port_to_odp_port(ntohs(opm->port_no)));
     if (!port) {
         return ofp_mkerr(OFPET_PORT_MOD_FAILED, OFPPMFC_BAD_PORT);
     } else if (memcmp(port->opp.hw_addr, opm->hw_addr, OFP_ETH_ALEN)) {
@@ -2916,19 +3086,19 @@ handle_table_stats_request(struct ofproto *p, struct ofconn *ofconn,
 }
 
 static void
-append_port_stat(struct ofport *port, uint16_t port_no, struct ofconn *ofconn, 
+append_port_stat(struct ofport *port, struct ofconn *ofconn,
                  struct ofpbuf **msgp)
 {
     struct netdev_stats stats;
     struct ofp_port_stats *ops;
 
-    /* Intentionally ignore return value, since errors will set 
-     * 'stats' to all-1s, which is correct for OpenFlow, and 
+    /* Intentionally ignore return value, since errors will set
+     * 'stats' to all-1s, which is correct for OpenFlow, and
      * netdev_get_stats() will log errors. */
     netdev_get_stats(port->netdev, &stats);
 
     ops = append_stats_reply(sizeof *ops, ofconn, msgp);
-    ops->port_no = htons(odp_port_to_ofp_port(port_no));
+    ops->port_no = htons(port->opp.port_no);
     memset(ops->pad, 0, sizeof ops->pad);
     ops->rx_packets = htonll(stats.rx_packets);
     ops->tx_packets = htonll(stats.tx_packets);
@@ -2953,7 +3123,6 @@ handle_port_stats_request(struct ofproto *p, struct ofconn *ofconn,
     struct ofp_port_stats *ops;
     struct ofpbuf *msg;
     struct ofport *port;
-    unsigned int port_no;
 
     if (arg_size != sizeof *psr) {
         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
@@ -2962,14 +3131,13 @@ handle_port_stats_request(struct ofproto *p, struct ofconn *ofconn,
 
     msg = start_stats_reply(osr, sizeof *ops * 16);
     if (psr->port_no != htons(OFPP_NONE)) {
-        port = port_array_get(&p->ports, 
-                ofp_port_to_odp_port(ntohs(psr->port_no)));
+        port = get_port(p, ofp_port_to_odp_port(ntohs(psr->port_no)));
         if (port) {
-            append_port_stat(port, ntohs(psr->port_no), ofconn, &msg);
+            append_port_stat(port, ofconn, &msg);
         }
     } else {
-        PORT_ARRAY_FOR_EACH (port, &p->ports, port_no) {
-            append_port_stat(port, port_no, ofconn, &msg);
+        HMAP_FOR_EACH (port, struct ofport, hmap_node, &p->ports) {
+            append_port_stat(port, ofconn, &msg);
         }
     }
 
@@ -3147,7 +3315,7 @@ flow_stats_ds_cb(struct cls_rule *rule_, void *cbdata_)
     ds_put_cstr(results, "\n");
 }
 
-/* Adds a pretty-printed description of all flows to 'results', including 
+/* Adds a pretty-printed description of all flows to 'results', including
  * those marked hidden by secchan (e.g., by in-band control). */
 void
 ofproto_get_all_flows(struct ofproto *p, struct ds *results)
@@ -3231,8 +3399,8 @@ handle_aggregate_stats_request(struct ofproto *p, struct ofconn *ofconn,
 
 struct queue_stats_cbdata {
     struct ofconn *ofconn;
+    struct ofport *ofport;
     struct ofpbuf *msg;
-    uint16_t port_no;
 };
 
 static void
@@ -3242,7 +3410,7 @@ put_queue_stats(struct queue_stats_cbdata *cbdata, uint32_t queue_id,
     struct ofp_queue_stats *reply;
 
     reply = append_stats_reply(sizeof *reply, cbdata->ofconn, &cbdata->msg);
-    reply->port_no = htons(cbdata->port_no);
+    reply->port_no = htons(cbdata->ofport->opp.port_no);
     memset(reply->pad, 0, sizeof reply->pad);
     reply->queue_id = htonl(queue_id);
     reply->tx_bytes = htonll(stats->tx_bytes);
@@ -3261,11 +3429,10 @@ handle_queue_stats_dump_cb(uint32_t queue_id,
 }
 
 static void
-handle_queue_stats_for_port(struct ofport *port, uint16_t port_no,
-                            uint32_t queue_id,
+handle_queue_stats_for_port(struct ofport *port, uint32_t queue_id,
                             struct queue_stats_cbdata *cbdata)
 {
-    cbdata->port_no = port_no;
+    cbdata->ofport = port;
     if (queue_id == OFPQ_ALL) {
         netdev_dump_queue_stats(port->netdev,
                                 handle_queue_stats_dump_cb, cbdata);
@@ -3301,13 +3468,13 @@ handle_queue_stats_request(struct ofproto *ofproto, struct ofconn *ofconn,
     port_no = ntohs(qsr->port_no);
     queue_id = ntohl(qsr->queue_id);
     if (port_no == OFPP_ALL) {
-        PORT_ARRAY_FOR_EACH (port, &ofproto->ports, port_no) {
-            handle_queue_stats_for_port(port, port_no, queue_id, &cbdata);
+        HMAP_FOR_EACH (port, struct ofport, hmap_node, &ofproto->ports) {
+            handle_queue_stats_for_port(port, queue_id, &cbdata);
         }
     } else if (port_no < ofproto->max_ports) {
-        port = port_array_get(&ofproto->ports, port_no);
+        port = get_port(ofproto, ofp_port_to_odp_port(port_no));
         if (port) {
-            handle_queue_stats_for_port(port, port_no, queue_id, &cbdata);
+            handle_queue_stats_for_port(port, queue_id, &cbdata);
         }
     } else {
         ofpbuf_delete(cbdata.msg);
@@ -3388,8 +3555,7 @@ update_stats(struct ofproto *ofproto, struct rule *rule,
         update_time(ofproto, rule, stats);
         rule->packet_count += stats->n_packets;
         rule->byte_count += stats->n_bytes;
-        netflow_flow_update_flags(&rule->nf_flow, stats->ip_tos,
-                                  stats->tcp_flags);
+        netflow_flow_update_flags(&rule->nf_flow, stats->tcp_flags);
     }
 }
 
@@ -3440,7 +3606,6 @@ add_flow(struct ofproto *p, struct ofconn *ofconn,
     }
 
     rule_insert(p, rule, packet, in_port);
-    ofpbuf_delete(packet);
     return error;
 }
 
@@ -3478,7 +3643,6 @@ send_buffered_packet(struct ofproto *ofproto, struct ofconn *ofconn,
 
     flow_extract(packet, 0, in_port, &flow);
     rule_execute(ofproto, rule, packet, &flow);
-    ofpbuf_delete(packet);
 
     return 0;
 }
@@ -3773,7 +3937,7 @@ handle_role_request(struct ofproto *ofproto,
     }
     nrr = (struct nx_role_request *) msg;
 
-    if (ofconn->type != OFCONN_CONTROLLER) {
+    if (ofconn->type != OFCONN_PRIMARY) {
         VLOG_WARN_RL(&rl, "ignoring role request on non-controller "
                      "connection");
         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_EPERM);
@@ -3955,7 +4119,7 @@ handle_odp_miss_msg(struct ofproto *p, struct ofpbuf *packet)
     rule = lookup_valid_rule(p, &flow);
     if (!rule) {
         /* Don't send a packet-in if OFPPC_NO_PACKET_IN asserted. */
-        struct ofport *port = port_array_get(&p->ports, msg->port);
+        struct ofport *port = get_port(p, msg->port);
         if (port) {
             if (port->opp.config & OFPPC_NO_PACKET_IN) {
                 COVERAGE_INC(ofproto_no_packet_in);
@@ -3986,9 +4150,6 @@ handle_odp_miss_msg(struct ofproto *p, struct ofpbuf *packet)
         }
     }
 
-    rule_execute(p, rule, &payload, &flow);
-    rule_reinstall(p, rule);
-
     if (rule->super && rule->super->cr.priority == FAIL_OPEN_PRIORITY) {
         /*
          * Extra-special case for fail-open mode.
@@ -4000,10 +4161,13 @@ handle_odp_miss_msg(struct ofproto *p, struct ofpbuf *packet)
          *
          * See the top-level comment in fail-open.c for more information.
          */
-        send_packet_in(p, packet);
-    } else {
-        ofpbuf_delete(packet);
+        send_packet_in(p, ofpbuf_clone_with_headroom(packet,
+                                                     DPIF_RECV_MSG_PADDING));
     }
+
+    ofpbuf_pull(packet, sizeof *msg);
+    rule_execute(p, rule, packet, &flow);
+    rule_reinstall(p, rule);
 }
 
 static void
@@ -4211,7 +4375,7 @@ active_timeout(struct ofproto *ofproto, struct rule *rule)
 
             if (odp_flow.stats.n_packets) {
                 update_time(ofproto, rule, &odp_flow.stats);
-                netflow_flow_update_flags(&rule->nf_flow, odp_flow.stats.ip_tos,
+                netflow_flow_update_flags(&rule->nf_flow,
                                           odp_flow.stats.tcp_flags);
             }
         }
@@ -4412,7 +4576,7 @@ pick_datapath_id(const struct ofproto *ofproto)
 {
     const struct ofport *port;
 
-    port = port_array_get(&ofproto->ports, ODPP_LOCAL);
+    port = get_port(ofproto, ODPP_LOCAL);
     if (port) {
         uint8_t ea[ETH_ADDR_LEN];
         int error;