ofproto/ofproto.c

   1 /*
   2  * Copyright (c) 2009, 2010 Nicira Networks.
   3  * Copyright (c) 2010 Jean Tourrilhes - HP-Labs.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at:
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17
  18 #include <config.h>
  19 #include "ofproto.h"
  20 #include <errno.h>
  21 #include <inttypes.h>
  22 #include <sys/socket.h>
  23 #include <net/if.h>
  24 #include <netinet/in.h>
  25 #include <stdbool.h>
  26 #include <stdlib.h>
  27 #include "classifier.h"
  28 #include "coverage.h"
  29 #include "discovery.h"
  30 #include "dynamic-string.h"
  31 #include "fail-open.h"
  32 #include "in-band.h"
  33 #include "mac-learning.h"
  34 #include "netdev.h"
  35 #include "netflow.h"
  36 #include "ofp-print.h"
  37 #include "ofp-util.h"
  38 #include "ofproto-sflow.h"
  39 #include "ofpbuf.h"
  40 #include "openflow/nicira-ext.h"
  41 #include "openflow/openflow.h"
  42 #include "openvswitch/xflow.h"
  43 #include "packets.h"
  44 #include "pinsched.h"
  45 #include "pktbuf.h"
  46 #include "poll-loop.h"
  47 #include "port-array.h"
  48 #include "rconn.h"
  49 #include "shash.h"
  50 #include "status.h"
  51 #include "stp.h"
  52 #include "stream-ssl.h"
  53 #include "svec.h"
  54 #include "tag.h"
  55 #include "timeval.h"
  56 #include "unixctl.h"
  57 #include "vconn.h"
  58 #include "vlog.h"
  59 #include "wdp.h"
  60 #include "xfif.h"
  61 #include "xtoxll.h"
  62
  63 VLOG_DEFINE_THIS_MODULE(ofproto)
  64
  65 #include "sflow_api.h"
  66
  67 struct ofproto_rule {
  68     uint64_t flow_cookie;       /* Controller-issued identifier.
  69                                    (Kept in network-byte order.) */
  70     bool send_flow_removed;     /* Send a flow removed message? */
  71     tag_type tags;              /* Tags (set only by hooks). */
  72 };
  73
  74 static struct ofproto_rule *
  75 ofproto_rule_cast(const struct wdp_rule *wdp_rule)
  76 {
  77     return wdp_rule->client_data;
  78 }
  79
  80 static void
  81 ofproto_rule_init(struct wdp_rule *wdp_rule)
  82 {
  83     wdp_rule->client_data = xzalloc(sizeof(struct ofproto_rule));
  84 }
  85
  86
  87 static inline bool
  88 rule_is_hidden(const struct wdp_rule *rule)
  89 {
  90     /* Rules with priority higher than UINT16_MAX are set up by ofproto itself
  91      * (e.g. by in-band control) and are intentionally hidden from the
  92      * controller. */
  93     if (rule->cr.flow.priority > UINT16_MAX) {
  94         return true;
  95     }
  96
  97     return false;
  98 }
  99
 100 static void delete_flow(struct ofproto *, struct wdp_rule *, uint8_t reason);
 101
 102 /* ofproto supports two kinds of OpenFlow connections:
 103  *
 104  *   - "Primary" connections to ordinary OpenFlow controllers.  ofproto
 105  *     maintains persistent connections to these controllers and by default
 106  *     sends them asynchronous messages such as packet-ins.
 107  *
 108  *   - "Service" connections, e.g. from ovs-ofctl.  When these connections
 109  *     drop, it is the other side's responsibility to reconnect them if
 110  *     necessary.  ofproto does not send them asynchronous messages by default.
 111  *
 112  * Currently, active (tcp, ssl, unix) connections are always "primary"
 113  * connections and passive (ptcp, pssl, punix) connections are always "service"
 114  * connections.  There is no inherent reason for this, but it reflects the
 115  * common case.
 116  */
 117 enum ofconn_type {
 118     OFCONN_PRIMARY,             /* An ordinary OpenFlow controller. */
 119     OFCONN_SERVICE              /* A service connection, e.g. "ovs-ofctl". */
 120 };
 121
 122 /* A listener for incoming OpenFlow "service" connections. */
 123 struct ofservice {
 124     struct hmap_node node;      /* In struct ofproto's "services" hmap. */
 125     struct pvconn *pvconn;      /* OpenFlow connection listener. */
 126
 127     /* These are not used by ofservice directly.  They are settings for
 128      * accepted "struct ofconn"s from the pvconn. */
 129     int probe_interval;         /* Max idle time before probing, in seconds. */
 130     int rate_limit;             /* Max packet-in rate in packets per second. */
 131     int burst_limit;            /* Limit on accumulating packet credits. */
 132 };
 133
 134 static struct ofservice *ofservice_lookup(struct ofproto *,
 135                                           const char *target);
 136 static int ofservice_create(struct ofproto *,
 137                             const struct ofproto_controller *);
 138 static void ofservice_reconfigure(struct ofservice *,
 139                                   const struct ofproto_controller *);
 140 static void ofservice_destroy(struct ofproto *, struct ofservice *);
 141
 142 /* An OpenFlow connection. */
 143 struct ofconn {
 144     struct ofproto *ofproto;    /* The ofproto that owns this connection. */
 145     struct list node;           /* In struct ofproto's "all_conns" list. */
 146     struct rconn *rconn;        /* OpenFlow connection. */
 147     enum ofconn_type type;      /* Type. */
 148     bool flow_mod_table_id;     /* NXT_FLOW_MOD_TABLE_ID enabled? */
 149
 150     /* OFPT_PACKET_IN related data. */
 151     struct rconn_packet_counter *packet_in_counter; /* # queued on 'rconn'. */
 152     struct pinsched *schedulers[2]; /* Indexed by reason code; see below. */
 153     struct pktbuf *pktbuf;         /* OpenFlow packet buffers. */
 154     int miss_send_len;             /* Bytes to send of buffered packets. */
 155
 156     /* Number of OpenFlow messages queued on 'rconn' as replies to OpenFlow
 157      * requests, and the maximum number before we stop reading OpenFlow
 158      * requests.  */
 159 #define OFCONN_REPLY_MAX 100
 160     struct rconn_packet_counter *reply_counter;
 161
 162     /* type == OFCONN_PRIMARY only. */
 163     enum nx_role role;           /* Role. */
 164     struct hmap_node hmap_node;  /* In struct ofproto's "controllers" map. */
 165     struct discovery *discovery; /* Controller discovery object, if enabled. */
 166     struct status_category *ss;  /* Switch status category. */
 167     enum ofproto_band band;      /* In-band or out-of-band? */
 168 };
 169
 170 /* We use OFPR_NO_MATCH and OFPR_ACTION as indexes into struct ofconn's
 171  * "schedulers" array.  Their values are 0 and 1, and their meanings and values
 172  * coincide with WDP_CHAN_MISS and WDP_CHAN_ACTION, so this is convenient.  In
 173  * case anything ever changes, check their values here.  */
 174 #define N_SCHEDULERS 2
 175 BUILD_ASSERT_DECL(OFPR_NO_MATCH == 0);
 176 BUILD_ASSERT_DECL(OFPR_NO_MATCH == WDP_CHAN_MISS);
 177 BUILD_ASSERT_DECL(OFPR_ACTION == 1);
 178 BUILD_ASSERT_DECL(OFPR_ACTION == WDP_CHAN_ACTION);
 179
 180 static struct ofconn *ofconn_create(struct ofproto *, struct rconn *,
 181                                     enum ofconn_type);
 182 static void ofconn_destroy(struct ofconn *);
 183 static void ofconn_run(struct ofconn *, struct ofproto *);
 184 static void ofconn_wait(struct ofconn *);
 185 static bool ofconn_receives_async_msgs(const struct ofconn *);
 186 static char *ofconn_make_name(const struct ofproto *, const char *target);
 187 static void ofconn_set_rate_limit(struct ofconn *, int rate, int burst);
 188
 189 static void queue_tx(struct ofpbuf *msg, const struct ofconn *ofconn,
 190                      struct rconn_packet_counter *counter);
 191
 192 static void send_packet_in(struct ofproto *, struct wdp_packet *);
 193 static void do_send_packet_in(struct wdp_packet *, void *ofconn);
 194
 195 struct ofproto {
 196     /* Settings. */
 197     uint64_t datapath_id;       /* Datapath ID. */
 198     uint64_t fallback_dpid;     /* Datapath ID if no better choice found. */
 199     char *mfr_desc;             /* Manufacturer. */
 200     char *hw_desc;              /* Hardware. */
 201     char *sw_desc;              /* Software version. */
 202     char *serial_desc;          /* Serial number. */
 203     char *dp_desc;              /* Datapath description. */
 204
 205     /* Datapath. */
 206     struct wdp *wdp;
 207     uint32_t max_ports;
 208
 209     /* Configuration. */
 210     struct switch_status *switch_status;
 211     struct fail_open *fail_open;
 212     struct netflow *netflow;
 213     struct ofproto_sflow *sflow;
 214     bool tun_id_from_cookie;    /* NXT_TUN_ID_FROM_COOKIE enabled? */
 215
 216     /* In-band control. */
 217     struct in_band *in_band;
 218     long long int next_in_band_update;
 219     struct sockaddr_in *extra_in_band_remotes;
 220     size_t n_extra_remotes;
 221
 222     /* OpenFlow connections. */
 223     struct hmap controllers;   /* Controller "struct ofconn"s. */
 224     struct list all_conns;     /* Contains "struct ofconn"s. */
 225     enum ofproto_fail_mode fail_mode;
 226
 227     /* OpenFlow listeners. */
 228     struct hmap services;       /* Contains "struct ofservice"s. */
 229     struct pvconn **snoops;
 230     size_t n_snoops;
 231 };
 232
 233 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
 234
 235 static uint64_t pick_datapath_id(const struct ofproto *);
 236 static uint64_t pick_fallback_dpid(void);
 237
 238 static void handle_wdp_packet(struct ofproto *, struct wdp_packet *);
 239
 240 static void handle_openflow(struct ofconn *, struct ofproto *,
 241                             struct ofpbuf *);
 242
 243 int
 244 ofproto_create(const char *datapath, const char *datapath_type,
 245                const struct ofhooks *ofhooks, void *aux,
 246                struct ofproto **ofprotop)
 247 {
 248     struct wdp_stats stats;
 249     struct ofproto *p;
 250     struct wdp *wdp;
 251     int error;
 252
 253     *ofprotop = NULL;
 254
 255     /* Connect to datapath and start listening for messages. */
 256     error = wdp_open(datapath, datapath_type, &wdp);
 257     if (error) {
 258         VLOG_ERR("failed to open datapath %s: %s", datapath, strerror(error));
 259         return error;
 260     }
 261     error = wdp_get_wdp_stats(wdp, &stats);
 262     if (error) {
 263         VLOG_ERR("failed to obtain stats for datapath %s: %s",
 264                  datapath, strerror(error));
 265         wdp_close(wdp);
 266         return error;
 267     }
 268     error = wdp_recv_set_mask(wdp, ((1 << WDP_CHAN_MISS)
 269                                     | (1 << WDP_CHAN_ACTION)
 270                                     | (1 << WDP_CHAN_SFLOW)));
 271     if (error) {
 272         VLOG_ERR("failed to listen on datapath %s: %s",
 273                  datapath, strerror(error));
 274         wdp_close(wdp);
 275         return error;
 276     }
 277     wdp_flow_flush(wdp);
 278     wdp_recv_purge(wdp);
 279     wdp_set_ofhooks(wdp, ofhooks, aux);
 280
 281     /* Initialize settings. */
 282     p = xzalloc(sizeof *p);
 283     p->fallback_dpid = pick_fallback_dpid();
 284     p->datapath_id = p->fallback_dpid;
 285     p->mfr_desc = xstrdup(DEFAULT_MFR_DESC);
 286     p->hw_desc = xstrdup(DEFAULT_HW_DESC);
 287     p->sw_desc = xstrdup(DEFAULT_SW_DESC);
 288     p->serial_desc = xstrdup(DEFAULT_SERIAL_DESC);
 289     p->dp_desc = xstrdup(DEFAULT_DP_DESC);
 290
 291     /* Initialize datapath. */
 292     p->wdp = wdp;
 293     p->max_ports = stats.max_ports;
 294
 295     /* Initialize submodules. */
 296     p->switch_status = switch_status_create(p);
 297     p->in_band = NULL;
 298     p->fail_open = NULL;
 299     p->netflow = NULL;
 300     p->sflow = NULL;
 301
 302     /* Initialize OpenFlow connections. */
 303     list_init(&p->all_conns);
 304     hmap_init(&p->controllers);
 305     hmap_init(&p->services);
 306     p->snoops = NULL;
 307     p->n_snoops = 0;
 308
 309     /* Pick final datapath ID. */
 310     p->datapath_id = pick_datapath_id(p);
 311     VLOG_INFO("using datapath ID %016"PRIx64, p->datapath_id);
 312
 313     *ofprotop = p;
 314     return 0;
 315 }
 316
 317 void
 318 ofproto_set_datapath_id(struct ofproto *p, uint64_t datapath_id)
 319 {
 320     uint64_t old_dpid = p->datapath_id;
 321     p->datapath_id = datapath_id ? datapath_id : pick_datapath_id(p);
 322     if (p->datapath_id != old_dpid) {
 323         VLOG_INFO("datapath ID changed to %016"PRIx64, p->datapath_id);
 324
 325         /* Force all active connections to reconnect, since there is no way to
 326          * notify a controller that the datapath ID has changed. */
 327         ofproto_reconnect_controllers(p);
 328     }
 329 }
 330
 331 static bool
 332 is_discovery_controller(const struct ofproto_controller *c)
 333 {
 334     return !strcmp(c->target, "discover");
 335 }
 336
 337 static bool
 338 is_in_band_controller(const struct ofproto_controller *c)
 339 {
 340     return is_discovery_controller(c) || c->band == OFPROTO_IN_BAND;
 341 }
 342
 343 /* Creates a new controller in 'ofproto'.  Some of the settings are initially
 344  * drawn from 'c', but update_controller() needs to be called later to finish
 345  * the new ofconn's configuration. */
 346 static void
 347 add_controller(struct ofproto *ofproto, const struct ofproto_controller *c)
 348 {
 349     struct discovery *discovery;
 350     struct ofconn *ofconn;
 351
 352     if (is_discovery_controller(c)) {
 353         int error = discovery_create(c->accept_re, c->update_resolv_conf,
 354                                      ofproto->wdp, ofproto->switch_status,
 355                                      &discovery);
 356         if (error) {
 357             return;
 358         }
 359     } else {
 360         discovery = NULL;
 361     }
 362
 363     ofconn = ofconn_create(ofproto, rconn_create(5, 8), OFCONN_PRIMARY);
 364     ofconn->pktbuf = pktbuf_create();
 365     ofconn->miss_send_len = OFP_DEFAULT_MISS_SEND_LEN;
 366     if (discovery) {
 367         ofconn->discovery = discovery;
 368     } else {
 369         char *name = ofconn_make_name(ofproto, c->target);
 370         rconn_connect(ofconn->rconn, c->target, name);
 371         free(name);
 372     }
 373     hmap_insert(&ofproto->controllers, &ofconn->hmap_node,
 374                 hash_string(c->target, 0));
 375 }
 376
 377 /* Reconfigures 'ofconn' to match 'c'.  This function cannot update an ofconn's
 378  * target or turn discovery on or off (these are done by creating new ofconns
 379  * and deleting old ones), but it can update the rest of an ofconn's
 380  * settings. */
 381 static void
 382 update_controller(struct ofconn *ofconn, const struct ofproto_controller *c)
 383 {
 384     int probe_interval;
 385
 386     ofconn->band = (is_in_band_controller(c)
 387                     ? OFPROTO_IN_BAND : OFPROTO_OUT_OF_BAND);
 388
 389     rconn_set_max_backoff(ofconn->rconn, c->max_backoff);
 390
 391     probe_interval = c->probe_interval ? MAX(c->probe_interval, 5) : 0;
 392     rconn_set_probe_interval(ofconn->rconn, probe_interval);
 393
 394     if (ofconn->discovery) {
 395         discovery_set_update_resolv_conf(ofconn->discovery,
 396                                          c->update_resolv_conf);
 397         discovery_set_accept_controller_re(ofconn->discovery, c->accept_re);
 398     }
 399
 400     ofconn_set_rate_limit(ofconn, c->rate_limit, c->burst_limit);
 401 }
 402
 403 static const char *
 404 ofconn_get_target(const struct ofconn *ofconn)
 405 {
 406     return ofconn->discovery ? "discover" : rconn_get_target(ofconn->rconn);
 407 }
 408
 409 static struct ofconn *
 410 find_controller_by_target(struct ofproto *ofproto, const char *target)
 411 {
 412     struct ofconn *ofconn;
 413
 414     HMAP_FOR_EACH_WITH_HASH (ofconn, struct ofconn, hmap_node,
 415                              hash_string(target, 0), &ofproto->controllers) {
 416         if (!strcmp(ofconn_get_target(ofconn), target)) {
 417             return ofconn;
 418         }
 419     }
 420     return NULL;
 421 }
 422
 423 static void
 424 update_in_band_remotes(struct ofproto *ofproto)
 425 {
 426     const struct ofconn *ofconn;
 427     struct sockaddr_in *addrs;
 428     size_t max_addrs, n_addrs;
 429     bool discovery;
 430     size_t i;
 431
 432     /* Allocate enough memory for as many remotes as we could possibly have. */
 433     max_addrs = ofproto->n_extra_remotes + hmap_count(&ofproto->controllers);
 434     addrs = xmalloc(max_addrs * sizeof *addrs);
 435     n_addrs = 0;
 436
 437     /* Add all the remotes. */
 438     discovery = false;
 439     HMAP_FOR_EACH (ofconn, struct ofconn, hmap_node, &ofproto->controllers) {
 440         struct sockaddr_in *sin = &addrs[n_addrs];
 441
 442         if (ofconn->band == OFPROTO_OUT_OF_BAND) {
 443             continue;
 444         }
 445
 446         sin->sin_addr.s_addr = rconn_get_remote_ip(ofconn->rconn);
 447         if (sin->sin_addr.s_addr) {
 448             sin->sin_port = rconn_get_remote_port(ofconn->rconn);
 449             n_addrs++;
 450         }
 451         if (ofconn->discovery) {
 452             discovery = true;
 453         }
 454     }
 455     for (i = 0; i < ofproto->n_extra_remotes; i++) {
 456         addrs[n_addrs++] = ofproto->extra_in_band_remotes[i];
 457     }
 458
 459     /* Create or update or destroy in-band.
 460      *
 461      * Ordinarily we only enable in-band if there's at least one remote
 462      * address, but discovery needs the in-band rules for DHCP to be installed
 463      * even before we know any remote addresses. */
 464     if (n_addrs || discovery) {
 465         if (!ofproto->in_band) {
 466             in_band_create(ofproto, ofproto->wdp, ofproto->switch_status,
 467                            &ofproto->in_band);
 468         }
 469         if (ofproto->in_band) {
 470             in_band_set_remotes(ofproto->in_band, addrs, n_addrs);
 471         }
 472         ofproto->next_in_band_update = time_msec() + 1000;
 473     } else {
 474         in_band_destroy(ofproto->in_band);
 475         ofproto->in_band = NULL;
 476     }
 477
 478     /* Clean up. */
 479     free(addrs);
 480 }
 481
 482 static void
 483 update_fail_open(struct ofproto *p)
 484 {
 485     struct ofconn *ofconn;
 486
 487     if (!hmap_is_empty(&p->controllers)
 488             && p->fail_mode == OFPROTO_FAIL_STANDALONE) {
 489         struct rconn **rconns;
 490         size_t n;
 491
 492         if (!p->fail_open) {
 493             p->fail_open = fail_open_create(p, p->switch_status);
 494         }
 495
 496         n = 0;
 497         rconns = xmalloc(hmap_count(&p->controllers) * sizeof *rconns);
 498         HMAP_FOR_EACH (ofconn, struct ofconn, hmap_node, &p->controllers) {
 499             rconns[n++] = ofconn->rconn;
 500         }
 501
 502         fail_open_set_controllers(p->fail_open, rconns, n);
 503         /* p->fail_open takes ownership of 'rconns'. */
 504     } else {
 505         fail_open_destroy(p->fail_open);
 506         p->fail_open = NULL;
 507     }
 508 }
 509
 510 void
 511 ofproto_set_controllers(struct ofproto *p,
 512                         const struct ofproto_controller *controllers,
 513                         size_t n_controllers)
 514 {
 515     struct shash new_controllers;
 516     struct ofconn *ofconn, *next_ofconn;
 517     struct ofservice *ofservice, *next_ofservice;
 518     bool ss_exists;
 519     size_t i;
 520
 521     /* Create newly configured controllers and services.
 522      * Create a name to ofproto_controller mapping in 'new_controllers'. */
 523     shash_init(&new_controllers);
 524     for (i = 0; i < n_controllers; i++) {
 525         const struct ofproto_controller *c = &controllers[i];
 526
 527         if (!vconn_verify_name(c->target) || !strcmp(c->target, "discover")) {
 528             if (!find_controller_by_target(p, c->target)) {
 529                 add_controller(p, c);
 530             }
 531         } else if (!pvconn_verify_name(c->target)) {
 532             if (!ofservice_lookup(p, c->target) && ofservice_create(p, c)) {
 533                 continue;
 534             }
 535         } else {
 536             VLOG_WARN_RL(&rl, "%s: unsupported controller \"%s\"",
 537                          wdp_name(p->wdp), c->target);
 538             continue;
 539         }
 540
 541         shash_add_once(&new_controllers, c->target, &controllers[i]);
 542     }
 543
 544     /* Delete controllers that are no longer configured.
 545      * Update configuration of all now-existing controllers. */
 546     ss_exists = false;
 547     HMAP_FOR_EACH_SAFE (ofconn, next_ofconn, struct ofconn, hmap_node,
 548                         &p->controllers) {
 549         struct ofproto_controller *c;
 550
 551         c = shash_find_data(&new_controllers, ofconn_get_target(ofconn));
 552         if (!c) {
 553             ofconn_destroy(ofconn);
 554         } else {
 555             update_controller(ofconn, c);
 556             if (ofconn->ss) {
 557                 ss_exists = true;
 558             }
 559         }
 560     }
 561
 562     /* Delete services that are no longer configured.
 563      * Update configuration of all now-existing services. */
 564     HMAP_FOR_EACH_SAFE (ofservice, next_ofservice, struct ofservice, node,
 565                         &p->services) {
 566         struct ofproto_controller *c;
 567
 568         c = shash_find_data(&new_controllers,
 569                             pvconn_get_name(ofservice->pvconn));
 570         if (!c) {
 571             ofservice_destroy(p, ofservice);
 572         } else {
 573             ofservice_reconfigure(ofservice, c);
 574         }
 575     }
 576
 577     shash_destroy(&new_controllers);
 578
 579     update_in_band_remotes(p);
 580     update_fail_open(p);
 581
 582     if (!hmap_is_empty(&p->controllers) && !ss_exists) {
 583         ofconn = CONTAINER_OF(hmap_first(&p->controllers),
 584                               struct ofconn, hmap_node);
 585         ofconn->ss = switch_status_register(p->switch_status, "remote",
 586                                             rconn_status_cb, ofconn->rconn);
 587     }
 588 }
 589
 590 void
 591 ofproto_set_fail_mode(struct ofproto *p, enum ofproto_fail_mode fail_mode)
 592 {
 593     p->fail_mode = fail_mode;
 594     update_fail_open(p);
 595 }
 596
 597 /* Drops the connections between 'ofproto' and all of its controllers, forcing
 598  * them to reconnect. */
 599 void
 600 ofproto_reconnect_controllers(struct ofproto *ofproto)
 601 {
 602     struct ofconn *ofconn;
 603
 604     LIST_FOR_EACH (ofconn, struct ofconn, node, &ofproto->all_conns) {
 605         rconn_reconnect(ofconn->rconn);
 606     }
 607 }
 608
 609 static bool
 610 any_extras_changed(const struct ofproto *ofproto,
 611                    const struct sockaddr_in *extras, size_t n)
 612 {
 613     size_t i;
 614
 615     if (n != ofproto->n_extra_remotes) {
 616         return true;
 617     }
 618
 619     for (i = 0; i < n; i++) {
 620         const struct sockaddr_in *old = &ofproto->extra_in_band_remotes[i];
 621         const struct sockaddr_in *new = &extras[i];
 622
 623         if (old->sin_addr.s_addr != new->sin_addr.s_addr ||
 624             old->sin_port != new->sin_port) {
 625             return true;
 626         }
 627     }
 628
 629     return false;
 630 }
 631
 632 /* Sets the 'n' TCP port addresses in 'extras' as ones to which 'ofproto''s
 633  * in-band control should guarantee access, in the same way that in-band
 634  * control guarantees access to OpenFlow controllers. */
 635 void
 636 ofproto_set_extra_in_band_remotes(struct ofproto *ofproto,
 637                                   const struct sockaddr_in *extras, size_t n)
 638 {
 639     if (!any_extras_changed(ofproto, extras, n)) {
 640         return;
 641     }
 642
 643     free(ofproto->extra_in_band_remotes);
 644     ofproto->n_extra_remotes = n;
 645     ofproto->extra_in_band_remotes = xmemdup(extras, n * sizeof *extras);
 646
 647     update_in_band_remotes(ofproto);
 648 }
 649
 650 void
 651 ofproto_set_desc(struct ofproto *p,
 652                  const char *mfr_desc, const char *hw_desc,
 653                  const char *sw_desc, const char *serial_desc,
 654                  const char *dp_desc)
 655 {
 656     struct ofp_desc_stats *ods;
 657
 658     if (mfr_desc) {
 659         if (strlen(mfr_desc) >= sizeof ods->mfr_desc) {
 660             VLOG_WARN("truncating mfr_desc, must be less than %zu characters",
 661                     sizeof ods->mfr_desc);
 662         }
 663         free(p->mfr_desc);
 664         p->mfr_desc = xstrdup(mfr_desc);
 665     }
 666     if (hw_desc) {
 667         if (strlen(hw_desc) >= sizeof ods->hw_desc) {
 668             VLOG_WARN("truncating hw_desc, must be less than %zu characters",
 669                     sizeof ods->hw_desc);
 670         }
 671         free(p->hw_desc);
 672         p->hw_desc = xstrdup(hw_desc);
 673     }
 674     if (sw_desc) {
 675         if (strlen(sw_desc) >= sizeof ods->sw_desc) {
 676             VLOG_WARN("truncating sw_desc, must be less than %zu characters",
 677                     sizeof ods->sw_desc);
 678         }
 679         free(p->sw_desc);
 680         p->sw_desc = xstrdup(sw_desc);
 681     }
 682     if (serial_desc) {
 683         if (strlen(serial_desc) >= sizeof ods->serial_num) {
 684             VLOG_WARN("truncating serial_desc, must be less than %zu "
 685                     "characters",
 686                     sizeof ods->serial_num);
 687         }
 688         free(p->serial_desc);
 689         p->serial_desc = xstrdup(serial_desc);
 690     }
 691     if (dp_desc) {
 692         if (strlen(dp_desc) >= sizeof ods->dp_desc) {
 693             VLOG_WARN("truncating dp_desc, must be less than %zu characters",
 694                     sizeof ods->dp_desc);
 695         }
 696         free(p->dp_desc);
 697         p->dp_desc = xstrdup(dp_desc);
 698     }
 699 }
 700
 701 static int
 702 set_pvconns(struct pvconn ***pvconnsp, size_t *n_pvconnsp,
 703             const struct svec *svec)
 704 {
 705     struct pvconn **pvconns = *pvconnsp;
 706     size_t n_pvconns = *n_pvconnsp;
 707     int retval = 0;
 708     size_t i;
 709
 710     for (i = 0; i < n_pvconns; i++) {
 711         pvconn_close(pvconns[i]);
 712     }
 713     free(pvconns);
 714
 715     pvconns = xmalloc(svec->n * sizeof *pvconns);
 716     n_pvconns = 0;
 717     for (i = 0; i < svec->n; i++) {
 718         const char *name = svec->names[i];
 719         struct pvconn *pvconn;
 720         int error;
 721
 722         error = pvconn_open(name, &pvconn);
 723         if (!error) {
 724             pvconns[n_pvconns++] = pvconn;
 725         } else {
 726             VLOG_ERR("failed to listen on %s: %s", name, strerror(error));
 727             if (!retval) {
 728                 retval = error;
 729             }
 730         }
 731     }
 732
 733     *pvconnsp = pvconns;
 734     *n_pvconnsp = n_pvconns;
 735
 736     return retval;
 737 }
 738
 739 int
 740 ofproto_set_snoops(struct ofproto *ofproto, const struct svec *snoops)
 741 {
 742     return set_pvconns(&ofproto->snoops, &ofproto->n_snoops, snoops);
 743 }
 744
 745 int
 746 ofproto_set_netflow(struct ofproto *ofproto,
 747                     const struct netflow_options *nf_options)
 748 {
 749     if (nf_options && nf_options->collectors.n) {
 750         if (!ofproto->netflow) {
 751             ofproto->netflow = netflow_create();
 752         }
 753         return netflow_set_options(ofproto->netflow, nf_options);
 754     } else {
 755         netflow_destroy(ofproto->netflow);
 756         ofproto->netflow = NULL;
 757         return 0;
 758     }
 759 }
 760
 761 void
 762 ofproto_set_sflow(struct ofproto *ofproto,
 763                   const struct ofproto_sflow_options *oso)
 764 {
 765     struct ofproto_sflow *os = ofproto->sflow;
 766     if (oso) {
 767         if (!os) {
 768             os = ofproto->sflow = ofproto_sflow_create(ofproto->wdp);
 769             /* XXX ofport */
 770         }
 771         ofproto_sflow_set_options(os, oso);
 772     } else {
 773         ofproto_sflow_destroy(os);
 774         ofproto->sflow = NULL;
 775     }
 776 }
 777
 778 int
 779 ofproto_set_stp(struct ofproto *ofproto OVS_UNUSED, bool enable_stp)
 780 {
 781     /* XXX */
 782     if (enable_stp) {
 783         VLOG_WARN("STP is not yet implemented");
 784         return EINVAL;
 785     } else {
 786         return 0;
 787     }
 788 }
 789
 790 uint64_t
 791 ofproto_get_datapath_id(const struct ofproto *ofproto)
 792 {
 793     return ofproto->datapath_id;
 794 }
 795
 796 bool
 797 ofproto_has_primary_controller(const struct ofproto *ofproto)
 798 {
 799     return !hmap_is_empty(&ofproto->controllers);
 800 }
 801
 802 enum ofproto_fail_mode
 803 ofproto_get_fail_mode(const struct ofproto *p)
 804 {
 805     return p->fail_mode;
 806 }
 807
 808 void
 809 ofproto_get_snoops(const struct ofproto *ofproto, struct svec *snoops)
 810 {
 811     size_t i;
 812
 813     for (i = 0; i < ofproto->n_snoops; i++) {
 814         svec_add(snoops, pvconn_get_name(ofproto->snoops[i]));
 815     }
 816 }
 817
 818 void
 819 ofproto_destroy(struct ofproto *p)
 820 {
 821     struct ofservice *ofservice, *next_ofservice;
 822     struct ofconn *ofconn, *next_ofconn;
 823     size_t i;
 824
 825     if (!p) {
 826         return;
 827     }
 828
 829     /* Destroy fail-open and in-band early, since they touch the classifier. */
 830     fail_open_destroy(p->fail_open);
 831     p->fail_open = NULL;
 832
 833     in_band_destroy(p->in_band);
 834     p->in_band = NULL;
 835     free(p->extra_in_band_remotes);
 836
 837     ofproto_flush_flows(p);
 838
 839     LIST_FOR_EACH_SAFE (ofconn, next_ofconn, struct ofconn, node,
 840                         &p->all_conns) {
 841         ofconn_destroy(ofconn);
 842     }
 843     hmap_destroy(&p->controllers);
 844
 845     wdp_close(p->wdp);
 846
 847     switch_status_destroy(p->switch_status);
 848     netflow_destroy(p->netflow);
 849     ofproto_sflow_destroy(p->sflow);
 850
 851     HMAP_FOR_EACH_SAFE (ofservice, next_ofservice, struct ofservice, node,
 852                         &p->services) {
 853         ofservice_destroy(p, ofservice);
 854     }
 855     hmap_destroy(&p->services);
 856
 857     for (i = 0; i < p->n_snoops; i++) {
 858         pvconn_close(p->snoops[i]);
 859     }
 860     free(p->snoops);
 861
 862     free(p->mfr_desc);
 863     free(p->hw_desc);
 864     free(p->sw_desc);
 865     free(p->serial_desc);
 866     free(p->dp_desc);
 867
 868     free(p);
 869 }
 870
 871 int
 872 ofproto_run(struct ofproto *p)
 873 {
 874     int error = ofproto_run1(p);
 875     if (!error) {
 876         error = ofproto_run2(p, false);
 877     }
 878     return error;
 879 }
 880
 881 /* Returns a "preference level" for snooping 'ofconn'.  A higher return value
 882  * means that 'ofconn' is more interesting for monitoring than a lower return
 883  * value. */
 884 static int
 885 snoop_preference(const struct ofconn *ofconn)
 886 {
 887     switch (ofconn->role) {
 888     case NX_ROLE_MASTER:
 889         return 3;
 890     case NX_ROLE_OTHER:
 891         return 2;
 892     case NX_ROLE_SLAVE:
 893         return 1;
 894     default:
 895         /* Shouldn't happen. */
 896         return 0;
 897     }
 898 }
 899
 900 /* One of ofproto's "snoop" pvconns has accepted a new connection on 'vconn'.
 901  * Connects this vconn to a controller. */
 902 static void
 903 add_snooper(struct ofproto *ofproto, struct vconn *vconn)
 904 {
 905     struct ofconn *ofconn, *best;
 906
 907     /* Pick a controller for monitoring. */
 908     best = NULL;
 909     LIST_FOR_EACH (ofconn, struct ofconn, node, &ofproto->all_conns) {
 910         if (ofconn->type == OFCONN_PRIMARY
 911             && (!best || snoop_preference(ofconn) > snoop_preference(best))) {
 912             best = ofconn;
 913         }
 914     }
 915
 916     if (best) {
 917         rconn_add_monitor(best->rconn, vconn);
 918     } else {
 919         VLOG_INFO_RL(&rl, "no controller connection to snoop");
 920         vconn_close(vconn);
 921     }
 922 }
 923
 924 static void
 925 ofproto_port_poll_cb(const struct ofp_phy_port *opp, uint8_t reason,
 926                      void *ofproto_)
 927 {
 928     /* XXX Should limit the number of queued port status change messages. */
 929     struct ofproto *ofproto = ofproto_;
 930     struct ofconn *ofconn;
 931
 932     LIST_FOR_EACH (ofconn, struct ofconn, node, &ofproto->all_conns) {
 933         struct ofp_port_status *ops;
 934         struct ofpbuf *b;
 935
 936         if (!ofconn_receives_async_msgs(ofconn)) {
 937             continue;
 938         }
 939
 940         ops = make_openflow_xid(sizeof *ops, OFPT_PORT_STATUS, 0, &b);
 941         ops->reason = reason;
 942         ops->desc = *opp;
 943         hton_ofp_phy_port(&ops->desc);
 944         queue_tx(b, ofconn, NULL);
 945     }
 946 }
 947
 948 int
 949 ofproto_run1(struct ofproto *p)
 950 {
 951     struct ofconn *ofconn, *next_ofconn;
 952     struct ofservice *ofservice;
 953     int i;
 954
 955     for (i = 0; i < 50; i++) {
 956         struct wdp_packet packet;
 957         int error;
 958
 959         error = wdp_recv(p->wdp, &packet);
 960         if (error) {
 961             if (error == ENODEV) {
 962                 /* Someone destroyed the datapath behind our back.  The caller
 963                  * better destroy us and give up, because we're just going to
 964                  * spin from here on out. */
 965                 static struct vlog_rate_limit rl2 = VLOG_RATE_LIMIT_INIT(1, 5);
 966                 VLOG_ERR_RL(&rl2, "%s: datapath was destroyed externally",
 967                             wdp_name(p->wdp));
 968                 return ENODEV;
 969             }
 970             break;
 971         }
 972
 973         handle_wdp_packet(p, xmemdup(&packet, sizeof packet));
 974     }
 975
 976     wdp_port_poll(p->wdp, ofproto_port_poll_cb, p);
 977
 978     if (p->in_band) {
 979         if (time_msec() >= p->next_in_band_update) {
 980             update_in_band_remotes(p);
 981         }
 982         in_band_run(p->in_band);
 983     }
 984
 985     LIST_FOR_EACH_SAFE (ofconn, next_ofconn, struct ofconn, node,
 986                         &p->all_conns) {
 987         ofconn_run(ofconn, p);
 988     }
 989
 990     /* Fail-open maintenance.  Do this after processing the ofconns since
 991      * fail-open checks the status of the controller rconn. */
 992     if (p->fail_open) {
 993         fail_open_run(p->fail_open);
 994     }
 995
 996     HMAP_FOR_EACH (ofservice, struct ofservice, node, &p->services) {
 997         struct vconn *vconn;
 998         int retval;
 999
1000         retval = pvconn_accept(ofservice->pvconn, OFP_VERSION, &vconn);
1001         if (!retval) {
1002             struct ofconn *ofconn;
1003             struct rconn *rconn;
1004             char *name;
1005
1006             rconn = rconn_create(ofservice->probe_interval, 0);
1007             name = ofconn_make_name(p, vconn_get_name(vconn));
1008             rconn_connect_unreliably(rconn, vconn, name);
1009             free(name);
1010
1011             ofconn = ofconn_create(p, rconn, OFCONN_SERVICE);
1012             ofconn_set_rate_limit(ofconn, ofservice->rate_limit,
1013                                   ofservice->burst_limit);
1014         } else if (retval != EAGAIN) {
1015             VLOG_WARN_RL(&rl, "accept failed (%s)", strerror(retval));
1016         }
1017     }
1018
1019     for (i = 0; i < p->n_snoops; i++) {
1020         struct vconn *vconn;
1021         int retval;
1022
1023         retval = pvconn_accept(p->snoops[i], OFP_VERSION, &vconn);
1024         if (!retval) {
1025             add_snooper(p, vconn);
1026         } else if (retval != EAGAIN) {
1027             VLOG_WARN_RL(&rl, "accept failed (%s)", strerror(retval));
1028         }
1029     }
1030
1031     if (p->netflow) {
1032         netflow_run(p->netflow);
1033     }
1034     if (p->sflow) {
1035         ofproto_sflow_run(p->sflow);
1036     }
1037
1038     return 0;
1039 }
1040
1041 int
1042 ofproto_run2(struct ofproto *p OVS_UNUSED, bool revalidate_all OVS_UNUSED)
1043 {
1044     return 0;
1045 }
1046
1047 void
1048 ofproto_wait(struct ofproto *p)
1049 {
1050     struct ofservice *ofservice;
1051     struct ofconn *ofconn;
1052     size_t i;
1053
1054     wdp_recv_wait(p->wdp);
1055     wdp_port_poll_wait(p->wdp);
1056     LIST_FOR_EACH (ofconn, struct ofconn, node, &p->all_conns) {
1057         ofconn_wait(ofconn);
1058     }
1059     if (p->in_band) {
1060         poll_timer_wait_until(p->next_in_band_update);
1061         in_band_wait(p->in_band);
1062     }
1063     if (p->fail_open) {
1064         fail_open_wait(p->fail_open);
1065     }
1066     if (p->sflow) {
1067         ofproto_sflow_wait(p->sflow);
1068     }
1069     HMAP_FOR_EACH (ofservice, struct ofservice, node, &p->services) {
1070         pvconn_wait(ofservice->pvconn);
1071     }
1072     for (i = 0; i < p->n_snoops; i++) {
1073         pvconn_wait(p->snoops[i]);
1074     }
1075 }
1076
1077 void
1078 ofproto_revalidate(struct ofproto *ofproto, tag_type tag)
1079 {
1080     wdp_revalidate(ofproto->wdp, tag);
1081 }
1082
1083 void
1084 ofproto_revalidate_all(struct ofproto *ofproto)
1085 {
1086     wdp_revalidate_all(ofproto->wdp);
1087 }
1088
1089 bool
1090 ofproto_is_alive(const struct ofproto *p)
1091 {
1092     return !hmap_is_empty(&p->controllers);
1093 }
1094
1095 int
1096 ofproto_send_packet(struct ofproto *p, const flow_t *flow,
1097                     const union ofp_action *actions, size_t n_actions,
1098                     const struct ofpbuf *packet)
1099 {
1100     /* XXX Should we translate the wdp_execute() errno value into an OpenFlow
1101      * error code? */
1102     wdp_execute(p->wdp, flow->in_port, actions, n_actions, packet);
1103     return 0;
1104 }
1105
1106 /* Intended for used by ofproto clients and ofproto submodules to add flows to
1107  * the flow table. */
1108 void
1109 ofproto_add_flow(struct ofproto *p, const flow_t *flow,
1110                  const union ofp_action *actions, size_t n_actions,
1111                  int idle_timeout)
1112 {
1113     struct wdp_flow_put put;
1114     struct wdp_rule *rule;
1115
1116     put.flags = WDP_PUT_CREATE | WDP_PUT_MODIFY | WDP_PUT_ALL;
1117     put.flow = flow;
1118     put.actions = actions;
1119     put.n_actions = n_actions;
1120     put.idle_timeout = idle_timeout;
1121     put.hard_timeout = 0;
1122     put.ofp_table_id = 0xff;
1123
1124     if (!wdp_flow_put(p->wdp, &put, NULL, &rule)) {
1125         ofproto_rule_init(rule);
1126     }
1127 }
1128
1129 /* Intended for used by ofproto clients and ofproto submodules to delete flows
1130  * that they earlier added to the flow table. */
1131 void
1132 ofproto_delete_flow(struct ofproto *ofproto, const flow_t *flow)
1133 {
1134     struct wdp_rule *rule = wdp_flow_get(ofproto->wdp, flow, UINT_MAX);
1135     if (rule) {
1136         delete_flow(ofproto, rule, OFPRR_DELETE);
1137     }
1138 }
1139
1140 void
1141 ofproto_flush_flows(struct ofproto *ofproto)
1142 {
1143     COVERAGE_INC(ofproto_flush);
1144     wdp_flow_flush(ofproto->wdp);
1145     if (ofproto->in_band) {
1146         in_band_flushed(ofproto->in_band);
1147     }
1148     if (ofproto->fail_open) {
1149         fail_open_flushed(ofproto->fail_open);
1150     }
1151 }
1152 \f
1153 static struct ofconn *
1154 ofconn_create(struct ofproto *p, struct rconn *rconn, enum ofconn_type type)
1155 {
1156     struct ofconn *ofconn = xzalloc(sizeof *ofconn);
1157     ofconn->ofproto = p;
1158     list_push_back(&p->all_conns, &ofconn->node);
1159     ofconn->rconn = rconn;
1160     ofconn->type = type;
1161     ofconn->role = NX_ROLE_OTHER;
1162     ofconn->packet_in_counter = rconn_packet_counter_create ();
1163     ofconn->pktbuf = NULL;
1164     ofconn->miss_send_len = 0;
1165     ofconn->reply_counter = rconn_packet_counter_create ();
1166     return ofconn;
1167 }
1168
1169 static void
1170 ofconn_destroy(struct ofconn *ofconn)
1171 {
1172     if (ofconn->type == OFCONN_PRIMARY) {
1173         hmap_remove(&ofconn->ofproto->controllers, &ofconn->hmap_node);
1174     }
1175     discovery_destroy(ofconn->discovery);
1176
1177     list_remove(&ofconn->node);
1178     switch_status_unregister(ofconn->ss);
1179     rconn_destroy(ofconn->rconn);
1180     rconn_packet_counter_destroy(ofconn->packet_in_counter);
1181     rconn_packet_counter_destroy(ofconn->reply_counter);
1182     pktbuf_destroy(ofconn->pktbuf);
1183     free(ofconn);
1184 }
1185
1186 static void
1187 ofconn_run(struct ofconn *ofconn, struct ofproto *p)
1188 {
1189     int iteration;
1190     size_t i;
1191
1192     if (ofconn->discovery) {
1193         char *controller_name;
1194         if (rconn_is_connectivity_questionable(ofconn->rconn)) {
1195             discovery_question_connectivity(ofconn->discovery);
1196         }
1197         if (discovery_run(ofconn->discovery, &controller_name)) {
1198             if (controller_name) {
1199                 char *ofconn_name = ofconn_make_name(p, controller_name);
1200                 rconn_connect(ofconn->rconn, controller_name, ofconn_name);
1201                 free(ofconn_name);
1202             } else {
1203                 rconn_disconnect(ofconn->rconn);
1204             }
1205         }
1206     }
1207
1208     for (i = 0; i < N_SCHEDULERS; i++) {
1209         pinsched_run(ofconn->schedulers[i], do_send_packet_in, ofconn);
1210     }
1211
1212     rconn_run(ofconn->rconn);
1213
1214     if (rconn_packet_counter_read (ofconn->reply_counter) < OFCONN_REPLY_MAX) {
1215         /* Limit the number of iterations to prevent other tasks from
1216          * starving. */
1217         for (iteration = 0; iteration < 50; iteration++) {
1218             struct ofpbuf *of_msg = rconn_recv(ofconn->rconn);
1219             if (!of_msg) {
1220                 break;
1221             }
1222             if (p->fail_open) {
1223                 fail_open_maybe_recover(p->fail_open);
1224             }
1225             handle_openflow(ofconn, p, of_msg);
1226             ofpbuf_delete(of_msg);
1227         }
1228     }
1229
1230     if (!ofconn->discovery && !rconn_is_alive(ofconn->rconn)) {
1231         ofconn_destroy(ofconn);
1232     }
1233 }
1234
1235 static void
1236 ofconn_wait(struct ofconn *ofconn)
1237 {
1238     int i;
1239
1240     if (ofconn->discovery) {
1241         discovery_wait(ofconn->discovery);
1242     }
1243     for (i = 0; i < N_SCHEDULERS; i++) {
1244         pinsched_wait(ofconn->schedulers[i]);
1245     }
1246     rconn_run_wait(ofconn->rconn);
1247     if (rconn_packet_counter_read (ofconn->reply_counter) < OFCONN_REPLY_MAX) {
1248         rconn_recv_wait(ofconn->rconn);
1249     } else {
1250         COVERAGE_INC(ofproto_ofconn_stuck);
1251     }
1252 }
1253
1254 /* Returns true if 'ofconn' should receive asynchronous messages. */
1255 static bool
1256 ofconn_receives_async_msgs(const struct ofconn *ofconn)
1257 {
1258     if (ofconn->type == OFCONN_PRIMARY) {
1259         /* Primary controllers always get asynchronous messages unless they
1260          * have configured themselves as "slaves".  */
1261         return ofconn->role != NX_ROLE_SLAVE;
1262     } else {
1263         /* Service connections don't get asynchronous messages unless they have
1264          * explicitly asked for them by setting a nonzero miss send length. */
1265         return ofconn->miss_send_len > 0;
1266     }
1267 }
1268
1269 /* Returns a human-readable name for an OpenFlow connection between 'ofproto'
1270  * and 'target', suitable for use in log messages for identifying the
1271  * connection.
1272  *
1273  * The name is dynamically allocated.  The caller should free it (with free())
1274  * when it is no longer needed. */
1275 static char *
1276 ofconn_make_name(const struct ofproto *ofproto, const char *target)
1277 {
1278     return xasprintf("%s<->%s", wdp_base_name(ofproto->wdp), target);
1279 }
1280
1281 static void
1282 ofconn_set_rate_limit(struct ofconn *ofconn, int rate, int burst)
1283 {
1284     int i;
1285
1286     for (i = 0; i < N_SCHEDULERS; i++) {
1287         struct pinsched **s = &ofconn->schedulers[i];
1288
1289         if (rate > 0) {
1290             if (!*s) {
1291                 *s = pinsched_create(rate, burst,
1292                                      ofconn->ofproto->switch_status);
1293             } else {
1294                 pinsched_set_limits(*s, rate, burst);
1295             }
1296         } else {
1297             pinsched_destroy(*s);
1298             *s = NULL;
1299         }
1300     }
1301 }
1302 \f
1303 static void
1304 ofservice_reconfigure(struct ofservice *ofservice,
1305                       const struct ofproto_controller *c)
1306 {
1307     ofservice->probe_interval = c->probe_interval;
1308     ofservice->rate_limit = c->rate_limit;
1309     ofservice->burst_limit = c->burst_limit;
1310 }
1311
1312 /* Creates a new ofservice in 'ofproto'.  Returns 0 if successful, otherwise a
1313  * positive errno value. */
1314 static int
1315 ofservice_create(struct ofproto *ofproto, const struct ofproto_controller *c)
1316 {
1317     struct ofservice *ofservice;
1318     struct pvconn *pvconn;
1319     int error;
1320
1321     error = pvconn_open(c->target, &pvconn);
1322     if (error) {
1323         return error;
1324     }
1325
1326     ofservice = xzalloc(sizeof *ofservice);
1327     hmap_insert(&ofproto->services, &ofservice->node,
1328                 hash_string(c->target, 0));
1329     ofservice->pvconn = pvconn;
1330
1331     ofservice_reconfigure(ofservice, c);
1332
1333     return 0;
1334 }
1335
1336 static void
1337 ofservice_destroy(struct ofproto *ofproto, struct ofservice *ofservice)
1338 {
1339     hmap_remove(&ofproto->services, &ofservice->node);
1340     pvconn_close(ofservice->pvconn);
1341     free(ofservice);
1342 }
1343
1344 /* Finds and returns the ofservice within 'ofproto' that has the given
1345  * 'target', or a null pointer if none exists. */
1346 static struct ofservice *
1347 ofservice_lookup(struct ofproto *ofproto, const char *target)
1348 {
1349     struct ofservice *ofservice;
1350
1351     HMAP_FOR_EACH_WITH_HASH (ofservice, struct ofservice, node,
1352                              hash_string(target, 0), &ofproto->services) {
1353         if (!strcmp(pvconn_get_name(ofservice->pvconn), target)) {
1354             return ofservice;
1355         }
1356     }
1357     return NULL;
1358 }
1359 \f
1360 static bool
1361 rule_has_out_port(const struct wdp_rule *rule, uint16_t out_port)
1362 {
1363     const union ofp_action *oa;
1364     struct actions_iterator i;
1365
1366     if (out_port == htons(OFPP_NONE)) {
1367         return true;
1368     }
1369     for (oa = actions_first(&i, rule->actions, rule->n_actions); oa;
1370          oa = actions_next(&i)) {
1371         if (action_outputs_to_port(oa, out_port)) {
1372             return true;
1373         }
1374     }
1375     return false;
1376 }
1377 \f
1378 static void
1379 queue_tx(struct ofpbuf *msg, const struct ofconn *ofconn,
1380          struct rconn_packet_counter *counter)
1381 {
1382     update_openflow_length(msg);
1383     if (rconn_send(ofconn->rconn, msg, counter)) {
1384         ofpbuf_delete(msg);
1385     }
1386 }
1387
1388 static void
1389 send_error(const struct ofconn *ofconn, const struct ofp_header *oh,
1390            int error, const void *data, size_t len)
1391 {
1392     struct ofpbuf *buf;
1393     struct ofp_error_msg *oem;
1394
1395     if (!(error >> 16)) {
1396         VLOG_WARN_RL(&rl, "not sending bad error code %d to controller",
1397                      error);
1398         return;
1399     }
1400
1401     COVERAGE_INC(ofproto_error);
1402     oem = make_openflow_xid(len + sizeof *oem, OFPT_ERROR,
1403                             oh ? oh->xid : 0, &buf);
1404     oem->type = htons((unsigned int) error >> 16);
1405     oem->code = htons(error & 0xffff);
1406     memcpy(oem->data, data, len);
1407     queue_tx(buf, ofconn, ofconn->reply_counter);
1408 }
1409
1410 static void
1411 send_error_oh(const struct ofconn *ofconn, const struct ofp_header *oh,
1412               int error)
1413 {
1414     size_t oh_length = ntohs(oh->length);
1415     send_error(ofconn, oh, error, oh, MIN(oh_length, 64));
1416 }
1417
1418 static int
1419 handle_echo_request(struct ofconn *ofconn, struct ofp_header *oh)
1420 {
1421     struct ofp_header *rq = oh;
1422     queue_tx(make_echo_reply(rq), ofconn, ofconn->reply_counter);
1423     return 0;
1424 }
1425
1426 static int
1427 handle_features_request(struct ofproto *p, struct ofconn *ofconn,
1428                         struct ofp_header *oh)
1429 {
1430     struct ofpbuf *features;
1431     int error;
1432
1433     error = wdp_get_features(p->wdp, &features);
1434     if (!error) {
1435         struct ofp_switch_features *osf = features->data;
1436
1437         update_openflow_length(features);
1438         osf->header.version = OFP_VERSION;
1439         osf->header.type = OFPT_FEATURES_REPLY;
1440         osf->header.xid = oh->xid;
1441
1442         osf->datapath_id = htonll(p->datapath_id);
1443         osf->n_buffers = htonl(pktbuf_capacity());
1444         memset(osf->pad, 0, sizeof osf->pad);
1445
1446         /* Turn on capabilities implemented by ofproto. */
1447         osf->capabilities |= htonl(OFPC_FLOW_STATS | OFPC_TABLE_STATS |
1448                                    OFPC_PORT_STATS);
1449
1450         queue_tx(features, ofconn, ofconn->reply_counter);
1451     }
1452     return error;
1453 }
1454
1455 static int
1456 handle_get_config_request(struct ofproto *p, struct ofconn *ofconn,
1457                           struct ofp_header *oh)
1458 {
1459     struct ofpbuf *buf;
1460     struct ofp_switch_config *osc;
1461     uint16_t flags;
1462     bool drop_frags;
1463
1464     /* Figure out flags. */
1465     wdp_get_drop_frags(p->wdp, &drop_frags);
1466     flags = drop_frags ? OFPC_FRAG_DROP : OFPC_FRAG_NORMAL;
1467
1468     /* Send reply. */
1469     osc = make_openflow_xid(sizeof *osc, OFPT_GET_CONFIG_REPLY, oh->xid, &buf);
1470     osc->flags = htons(flags);
1471     osc->miss_send_len = htons(ofconn->miss_send_len);
1472     queue_tx(buf, ofconn, ofconn->reply_counter);
1473
1474     return 0;
1475 }
1476
1477 static int
1478 handle_set_config(struct ofproto *p, struct ofconn *ofconn,
1479                   struct ofp_switch_config *osc)
1480 {
1481     uint16_t flags;
1482     int error;
1483
1484     error = check_ofp_message(&osc->header, OFPT_SET_CONFIG, sizeof *osc);
1485     if (error) {
1486         return error;
1487     }
1488     flags = ntohs(osc->flags);
1489
1490     if (ofconn->type == OFCONN_PRIMARY && ofconn->role != NX_ROLE_SLAVE) {
1491         switch (flags & OFPC_FRAG_MASK) {
1492         case OFPC_FRAG_NORMAL:
1493             wdp_set_drop_frags(p->wdp, false);
1494             break;
1495         case OFPC_FRAG_DROP:
1496             wdp_set_drop_frags(p->wdp, true);
1497             break;
1498         default:
1499             VLOG_WARN_RL(&rl, "requested bad fragment mode (flags=%"PRIx16")",
1500                          osc->flags);
1501             break;
1502         }
1503     }
1504
1505     ofconn->miss_send_len = ntohs(osc->miss_send_len);
1506
1507     return 0;
1508 }
1509
1510 /* Checks whether 'ofconn' is a slave controller.  If so, returns an OpenFlow
1511  * error message code (composed with ofp_mkerr()) for the caller to propagate
1512  * upward.  Otherwise, returns 0.
1513  *
1514  * 'oh' is used to make log messages more informative. */
1515 static int
1516 reject_slave_controller(struct ofconn *ofconn, const struct ofp_header *oh)
1517 {
1518     if (ofconn->type == OFCONN_PRIMARY && ofconn->role == NX_ROLE_SLAVE) {
1519         static struct vlog_rate_limit perm_rl = VLOG_RATE_LIMIT_INIT(1, 5);
1520         char *type_name;
1521
1522         type_name = ofp_message_type_to_string(oh->type);
1523         VLOG_WARN_RL(&perm_rl, "rejecting %s message from slave controller",
1524                      type_name);
1525         free(type_name);
1526
1527         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_EPERM);
1528     } else {
1529         return 0;
1530     }
1531 }
1532
1533 static int
1534 handle_packet_out(struct ofproto *p, struct ofconn *ofconn,
1535                   struct ofp_header *oh)
1536 {
1537     struct ofp_packet_out *opo;
1538     struct ofpbuf payload, *buffer;
1539     struct ofp_action_header *actions;
1540     int n_actions;
1541     uint16_t in_port;
1542     flow_t flow;
1543     int error;
1544
1545     error = reject_slave_controller(ofconn, oh);
1546     if (error) {
1547         return error;
1548     }
1549
1550     error = check_ofp_packet_out(oh, &payload, &n_actions, p->max_ports);
1551     if (error) {
1552         return error;
1553     }
1554     opo = (struct ofp_packet_out *) oh;
1555     actions = opo->actions;
1556
1557     COVERAGE_INC(ofproto_packet_out);
1558     if (opo->buffer_id != htonl(UINT32_MAX)) {
1559         error = pktbuf_retrieve(ofconn->pktbuf, ntohl(opo->buffer_id),
1560                                 &buffer, &in_port);
1561         if (error || !buffer) {
1562             return error;
1563         }
1564         payload = *buffer;
1565     } else {
1566         buffer = NULL;
1567     }
1568
1569     flow_extract(&payload, 0, ntohs(opo->in_port), &flow);
1570     wdp_execute(p->wdp, flow.in_port, (const union ofp_action *) actions,
1571                 n_actions, &payload);
1572     ofpbuf_delete(buffer);
1573
1574     return 0;
1575 }
1576
1577 static int
1578 handle_port_mod(struct ofproto *p, struct ofconn *ofconn,
1579                 struct ofp_header *oh)
1580 {
1581     const struct ofp_port_mod *opm;
1582     struct wdp_port port;
1583     int error;
1584
1585     error = reject_slave_controller(ofconn, oh);
1586     if (error) {
1587         return error;
1588     }
1589     error = check_ofp_message(oh, OFPT_PORT_MOD, sizeof *opm);
1590     if (error) {
1591         return error;
1592     }
1593     opm = (struct ofp_port_mod *) oh;
1594
1595     if (wdp_port_query_by_number(p->wdp, ntohs(opm->port_no), &port)) {
1596         error = ofp_mkerr(OFPET_PORT_MOD_FAILED, OFPPMFC_BAD_PORT);
1597     } else if (memcmp(port.opp.hw_addr, opm->hw_addr, OFP_ETH_ALEN)) {
1598         error = ofp_mkerr(OFPET_PORT_MOD_FAILED, OFPPMFC_BAD_HW_ADDR);
1599     } else {
1600         uint32_t mask, new_config;
1601
1602         mask = ntohl(opm->mask) & (OFPPC_PORT_DOWN | OFPPC_NO_STP
1603                                    | OFPPC_NO_RECV | OFPPC_NO_RECV_STP
1604                                    | OFPPC_NO_FLOOD | OFPPC_NO_FWD
1605                                    | OFPPC_NO_PACKET_IN);
1606         new_config = (port.opp.config & ~mask) | (ntohl(opm->config) & mask);
1607         if (new_config != port.opp.config) {
1608             error = wdp_port_set_config(p->wdp, ntohs(opm->port_no),
1609                                         new_config);
1610         }
1611         if (opm->advertise) {
1612             netdev_set_advertisements(port.netdev, ntohl(opm->advertise));
1613         }
1614     }
1615     wdp_port_free(&port);
1616
1617     return error;
1618 }
1619
1620 static struct ofpbuf *
1621 make_stats_reply(uint32_t xid, uint16_t type, size_t body_len)
1622 {
1623     struct ofp_stats_reply *osr;
1624     struct ofpbuf *msg;
1625
1626     msg = ofpbuf_new(MIN(sizeof *osr + body_len, UINT16_MAX));
1627     osr = put_openflow_xid(sizeof *osr, OFPT_STATS_REPLY, xid, msg);
1628     osr->type = type;
1629     osr->flags = htons(0);
1630     return msg;
1631 }
1632
1633 static struct ofpbuf *
1634 start_stats_reply(const struct ofp_stats_request *request, size_t body_len)
1635 {
1636     return make_stats_reply(request->header.xid, request->type, body_len);
1637 }
1638
1639 static void *
1640 append_stats_reply(size_t nbytes, struct ofconn *ofconn, struct ofpbuf **msgp)
1641 {
1642     struct ofpbuf *msg = *msgp;
1643     assert(nbytes <= UINT16_MAX - sizeof(struct ofp_stats_reply));
1644     if (nbytes + msg->size > UINT16_MAX) {
1645         struct ofp_stats_reply *reply = msg->data;
1646         reply->flags = htons(OFPSF_REPLY_MORE);
1647         *msgp = make_stats_reply(reply->header.xid, reply->type, nbytes);
1648         queue_tx(msg, ofconn, ofconn->reply_counter);
1649     }
1650     return ofpbuf_put_uninit(*msgp, nbytes);
1651 }
1652
1653 static int
1654 handle_desc_stats_request(struct ofproto *p, struct ofconn *ofconn,
1655                            struct ofp_stats_request *request)
1656 {
1657     struct ofp_desc_stats *ods;
1658     struct ofpbuf *msg;
1659
1660     msg = start_stats_reply(request, sizeof *ods);
1661     ods = append_stats_reply(sizeof *ods, ofconn, &msg);
1662     memset(ods, 0, sizeof *ods);
1663     ovs_strlcpy(ods->mfr_desc, p->mfr_desc, sizeof ods->mfr_desc);
1664     ovs_strlcpy(ods->hw_desc, p->hw_desc, sizeof ods->hw_desc);
1665     ovs_strlcpy(ods->sw_desc, p->sw_desc, sizeof ods->sw_desc);
1666     ovs_strlcpy(ods->serial_num, p->serial_desc, sizeof ods->serial_num);
1667     ovs_strlcpy(ods->dp_desc, p->dp_desc, sizeof ods->dp_desc);
1668     queue_tx(msg, ofconn, ofconn->reply_counter);
1669
1670     return 0;
1671 }
1672
1673 static int
1674 handle_table_stats_request(struct ofproto *p, struct ofconn *ofconn,
1675                            struct ofp_stats_request *request)
1676 {
1677     struct ofpbuf *msg;
1678     int error;
1679
1680     msg = start_stats_reply(request, sizeof(struct ofp_table_stats) * 3);
1681     error = wdp_get_table_stats(p->wdp, msg);
1682     if (!error) {
1683         queue_tx(msg, ofconn, ofconn->reply_counter);
1684     } else {
1685         ofpbuf_delete(msg);
1686     }
1687     return error;
1688 }
1689
1690 static void
1691 append_port_stat(struct wdp_port *port, struct ofconn *ofconn,
1692                  struct ofpbuf **msgp)
1693 {
1694     struct netdev_stats stats;
1695     struct ofp_port_stats *ops;
1696
1697     /* Intentionally ignore return value, since errors will set
1698      * 'stats' to all-1s, which is correct for OpenFlow, and
1699      * netdev_get_stats() will log errors. */
1700     netdev_get_stats(port->netdev, &stats);
1701
1702     ops = append_stats_reply(sizeof *ops, ofconn, msgp);
1703     ops->port_no = htons(port->opp.port_no);
1704     memset(ops->pad, 0, sizeof ops->pad);
1705     ops->rx_packets = htonll(stats.rx_packets);
1706     ops->tx_packets = htonll(stats.tx_packets);
1707     ops->rx_bytes = htonll(stats.rx_bytes);
1708     ops->tx_bytes = htonll(stats.tx_bytes);
1709     ops->rx_dropped = htonll(stats.rx_dropped);
1710     ops->tx_dropped = htonll(stats.tx_dropped);
1711     ops->rx_errors = htonll(stats.rx_errors);
1712     ops->tx_errors = htonll(stats.tx_errors);
1713     ops->rx_frame_err = htonll(stats.rx_frame_errors);
1714     ops->rx_over_err = htonll(stats.rx_over_errors);
1715     ops->rx_crc_err = htonll(stats.rx_crc_errors);
1716     ops->collisions = htonll(stats.collisions);
1717 }
1718
1719 static int
1720 handle_port_stats_request(struct ofproto *p, struct ofconn *ofconn,
1721                           struct ofp_stats_request *osr,
1722                           size_t arg_size)
1723 {
1724     struct ofp_port_stats_request *psr;
1725     struct ofp_port_stats *ops;
1726     struct ofpbuf *msg;
1727
1728     if (arg_size != sizeof *psr) {
1729         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
1730     }
1731     psr = (struct ofp_port_stats_request *) osr->body;
1732
1733     msg = start_stats_reply(osr, sizeof *ops * 16);
1734     if (psr->port_no != htons(OFPP_NONE)) {
1735         struct wdp_port port;
1736
1737         if (!wdp_port_query_by_number(p->wdp, ntohs(psr->port_no), &port)) {
1738             append_port_stat(&port, ofconn, &msg);
1739             wdp_port_free(&port);
1740         }
1741     } else {
1742         struct wdp_port *ports;
1743         size_t n_ports;
1744         size_t i;
1745
1746         wdp_port_list(p->wdp, &ports, &n_ports);
1747         for (i = 0; i < n_ports; i++) {
1748             append_port_stat(&ports[i], ofconn, &msg);
1749         }
1750         wdp_port_array_free(ports, n_ports);
1751     }
1752
1753     queue_tx(msg, ofconn, ofconn->reply_counter);
1754     return 0;
1755 }
1756
1757 struct flow_stats_cbdata {
1758     struct ofproto *ofproto;
1759     struct ofconn *ofconn;
1760     uint16_t out_port;
1761     struct ofpbuf *msg;
1762 };
1763
1764 /* Obtains statistic counters for 'rule' within 'p' and stores them into
1765  * '*packet_countp' and '*byte_countp'.  If 'rule' is a wildcarded rule, the
1766  * returned statistic include statistics for all of 'rule''s subrules. */
1767 static void
1768 query_stats(struct ofproto *p, struct wdp_rule *rule,
1769             uint64_t *packet_countp, uint64_t *byte_countp)
1770 {
1771     struct wdp_flow_stats stats;
1772
1773     if (!wdp_flow_get_stats(p->wdp, rule, &stats)) {
1774         *packet_countp = stats.n_packets;
1775         *byte_countp = stats.n_bytes;
1776     } else {
1777         *packet_countp = 0;
1778         *byte_countp = 0;
1779     }
1780 }
1781
1782 static int
1783 flow_stats_cb(struct wdp_rule *rule, void *cbdata_)
1784 {
1785     struct flow_stats_cbdata *cbdata = cbdata_;
1786     struct ofp_flow_stats *ofs;
1787     uint64_t packet_count, byte_count;
1788     size_t act_len, len;
1789     long long int tdiff = time_msec() - rule->created;
1790     uint32_t sec = tdiff / 1000;
1791     uint32_t msec = tdiff - (sec * 1000);
1792
1793     if (rule_is_hidden(rule)
1794         || !rule_has_out_port(rule, cbdata->out_port)) {
1795         return 0;
1796     }
1797
1798     act_len = sizeof *rule->actions * rule->n_actions;
1799     len = offsetof(struct ofp_flow_stats, actions) + act_len;
1800
1801     query_stats(cbdata->ofproto, rule, &packet_count, &byte_count);
1802
1803     ofs = append_stats_reply(len, cbdata->ofconn, &cbdata->msg);
1804     ofs->length = htons(len);
1805     ofs->table_id = rule->ofp_table_id;
1806     ofs->pad = 0;
1807     flow_to_match(&rule->cr.flow, cbdata->ofproto->tun_id_from_cookie,
1808                   &ofs->match);
1809     ofs->duration_sec = htonl(sec);
1810     ofs->duration_nsec = htonl(msec * 1000000);
1811     ofs->cookie = ofproto_rule_cast(rule)->flow_cookie;
1812     ofs->priority = htons(rule->cr.flow.priority);
1813     ofs->idle_timeout = htons(rule->idle_timeout);
1814     ofs->hard_timeout = htons(rule->hard_timeout);
1815     memset(ofs->pad2, 0, sizeof ofs->pad2);
1816     ofs->packet_count = htonll(packet_count);
1817     ofs->byte_count = htonll(byte_count);
1818     memcpy(ofs->actions, rule->actions, act_len);
1819
1820     return 0;
1821 }
1822
1823 static unsigned int
1824 table_id_to_include(uint8_t table_id)
1825 {
1826     return (table_id == 0xff ? UINT_MAX
1827             : table_id < 32 ? 1u << table_id
1828             : 0);
1829 }
1830
1831 static unsigned int
1832 flow_mod_table_id(const struct ofconn *ofconn, const struct ofp_flow_mod *ofm)
1833 {
1834     return ofconn->flow_mod_table_id ? ntohs(ofm->command) >> 8 : 0xff;
1835 }
1836
1837 static int
1838 handle_flow_stats_request(struct ofproto *p, struct ofconn *ofconn,
1839                           const struct ofp_stats_request *osr,
1840                           size_t arg_size)
1841 {
1842     struct ofp_flow_stats_request *fsr;
1843     struct flow_stats_cbdata cbdata;
1844     flow_t target;
1845
1846     if (arg_size != sizeof *fsr) {
1847         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
1848     }
1849     fsr = (struct ofp_flow_stats_request *) osr->body;
1850
1851     COVERAGE_INC(ofproto_flows_req);
1852     cbdata.ofproto = p;
1853     cbdata.ofconn = ofconn;
1854     cbdata.out_port = fsr->out_port;
1855     cbdata.msg = start_stats_reply(osr, 1024);
1856     flow_from_match(&fsr->match, 0, false, 0, &target);
1857     wdp_flow_for_each_match(p->wdp, &target,
1858                             table_id_to_include(fsr->table_id),
1859                             flow_stats_cb, &cbdata);
1860     queue_tx(cbdata.msg, ofconn, ofconn->reply_counter);
1861     return 0;
1862 }
1863
1864 struct flow_stats_ds_cbdata {
1865     struct ofproto *ofproto;
1866     struct ds *results;
1867 };
1868
1869 static int
1870 flow_stats_ds_cb(struct wdp_rule *rule, void *cbdata_)
1871 {
1872     struct flow_stats_ds_cbdata *cbdata = cbdata_;
1873     struct ds *results = cbdata->results;
1874     struct ofp_match match;
1875     uint64_t packet_count, byte_count;
1876     size_t act_len = sizeof *rule->actions * rule->n_actions;
1877
1878     query_stats(cbdata->ofproto, rule, &packet_count, &byte_count);
1879     flow_to_match(&rule->cr.flow, cbdata->ofproto->tun_id_from_cookie,
1880                   &match);
1881
1882     ds_put_format(results, "duration=%llds, ",
1883                   (time_msec() - rule->created) / 1000);
1884     ds_put_format(results, "priority=%u, ", rule->cr.flow.priority);
1885     ds_put_format(results, "n_packets=%"PRIu64", ", packet_count);
1886     ds_put_format(results, "n_bytes=%"PRIu64", ", byte_count);
1887     ofp_print_match(results, &match, true);
1888     ofp_print_actions(results, &rule->actions->header, act_len);
1889     ds_put_cstr(results, "\n");
1890
1891     return 0;
1892 }
1893
1894 /* Adds a pretty-printed description of all flows to 'results', including
1895  * those marked hidden by secchan (e.g., by in-band control). */
1896 void
1897 ofproto_get_all_flows(struct ofproto *p, struct ds *results)
1898 {
1899     struct flow_stats_ds_cbdata cbdata;
1900     struct ofp_match match;
1901     flow_t target;
1902
1903     memset(&match, 0, sizeof match);
1904     match.wildcards = htonl(OVSFW_ALL);
1905
1906     cbdata.ofproto = p;
1907     cbdata.results = results;
1908
1909     flow_from_match(&match, 0, false, 0, &target);
1910     wdp_flow_for_each_match(p->wdp, &target, UINT_MAX,
1911                             flow_stats_ds_cb, &cbdata);
1912 }
1913
1914 struct aggregate_stats_cbdata {
1915     struct ofproto *ofproto;
1916     uint16_t out_port;
1917     uint64_t packet_count;
1918     uint64_t byte_count;
1919     uint32_t n_flows;
1920 };
1921
1922 static int
1923 aggregate_stats_cb(struct wdp_rule *rule, void *cbdata_)
1924 {
1925     struct aggregate_stats_cbdata *cbdata = cbdata_;
1926     uint64_t packet_count, byte_count;
1927
1928     if (rule_is_hidden(rule) || !rule_has_out_port(rule, cbdata->out_port)) {
1929         return 0;
1930     }
1931
1932     query_stats(cbdata->ofproto, rule, &packet_count, &byte_count);
1933
1934     cbdata->packet_count += packet_count;
1935     cbdata->byte_count += byte_count;
1936     cbdata->n_flows++;
1937
1938     return 0;
1939 }
1940
1941 static int
1942 handle_aggregate_stats_request(struct ofproto *p, struct ofconn *ofconn,
1943                                const struct ofp_stats_request *osr,
1944                                size_t arg_size)
1945 {
1946     struct ofp_aggregate_stats_request *asr;
1947     struct ofp_aggregate_stats_reply *reply;
1948     struct aggregate_stats_cbdata cbdata;
1949     struct ofpbuf *msg;
1950     flow_t target;
1951
1952     if (arg_size != sizeof *asr) {
1953         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
1954     }
1955     asr = (struct ofp_aggregate_stats_request *) osr->body;
1956
1957     COVERAGE_INC(ofproto_agg_request);
1958     cbdata.ofproto = p;
1959     cbdata.out_port = asr->out_port;
1960     cbdata.packet_count = 0;
1961     cbdata.byte_count = 0;
1962     cbdata.n_flows = 0;
1963     flow_from_match(&asr->match, 0, false, 0, &target);
1964     wdp_flow_for_each_match(p->wdp, &target,
1965                             table_id_to_include(asr->table_id),
1966                             aggregate_stats_cb, &cbdata);
1967
1968     msg = start_stats_reply(osr, sizeof *reply);
1969     reply = append_stats_reply(sizeof *reply, ofconn, &msg);
1970     reply->flow_count = htonl(cbdata.n_flows);
1971     reply->packet_count = htonll(cbdata.packet_count);
1972     reply->byte_count = htonll(cbdata.byte_count);
1973     queue_tx(msg, ofconn, ofconn->reply_counter);
1974     return 0;
1975 }
1976
1977 struct queue_stats_cbdata {
1978     struct ofconn *ofconn;
1979     struct ofpbuf *msg;
1980     uint16_t port_no;
1981 };
1982
1983 static void
1984 put_queue_stats(struct queue_stats_cbdata *cbdata, uint32_t queue_id,
1985                 const struct netdev_queue_stats *stats)
1986 {
1987     struct ofp_queue_stats *reply;
1988
1989     reply = append_stats_reply(sizeof *reply, cbdata->ofconn, &cbdata->msg);
1990     reply->port_no = htons(cbdata->port_no);
1991     memset(reply->pad, 0, sizeof reply->pad);
1992     reply->queue_id = htonl(queue_id);
1993     reply->tx_bytes = htonll(stats->tx_bytes);
1994     reply->tx_packets = htonll(stats->tx_packets);
1995     reply->tx_errors = htonll(stats->tx_errors);
1996 }
1997
1998 static void
1999 handle_queue_stats_dump_cb(uint32_t queue_id,
2000                            struct netdev_queue_stats *stats,
2001                            void *cbdata_)
2002 {
2003     struct queue_stats_cbdata *cbdata = cbdata_;
2004
2005     put_queue_stats(cbdata, queue_id, stats);
2006 }
2007
2008 static void
2009 handle_queue_stats_for_port(struct wdp_port *port, uint32_t queue_id,
2010                             struct queue_stats_cbdata *cbdata)
2011 {
2012     cbdata->port_no = port->opp.port_no;
2013     if (queue_id == OFPQ_ALL) {
2014         netdev_dump_queue_stats(port->netdev,
2015                                 handle_queue_stats_dump_cb, cbdata);
2016     } else {
2017         struct netdev_queue_stats stats;
2018
2019         netdev_get_queue_stats(port->netdev, queue_id, &stats);
2020         put_queue_stats(cbdata, queue_id, &stats);
2021     }
2022 }
2023
2024 static int
2025 handle_queue_stats_request(struct ofproto *ofproto, struct ofconn *ofconn,
2026                            const struct ofp_stats_request *osr,
2027                            size_t arg_size)
2028 {
2029     struct ofp_queue_stats_request *qsr;
2030     struct queue_stats_cbdata cbdata;
2031     unsigned int port_no;
2032     uint32_t queue_id;
2033
2034     if (arg_size != sizeof *qsr) {
2035         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
2036     }
2037     qsr = (struct ofp_queue_stats_request *) osr->body;
2038
2039     COVERAGE_INC(ofproto_queue_req);
2040
2041     cbdata.ofconn = ofconn;
2042     cbdata.msg = start_stats_reply(osr, 128);
2043
2044     port_no = ntohs(qsr->port_no);
2045     queue_id = ntohl(qsr->queue_id);
2046     if (port_no == OFPP_ALL) {
2047         struct wdp_port *ports;
2048         size_t n_ports, i;
2049
2050         wdp_port_list(ofproto->wdp, &ports, &n_ports);
2051         /* XXX deal with wdp_port_list() errors */
2052         for (i = 0; i < n_ports; i++) {
2053             handle_queue_stats_for_port(&ports[i], queue_id, &cbdata);
2054         }
2055         wdp_port_array_free(ports, n_ports);
2056     } else if (port_no < ofproto->max_ports) {
2057         struct wdp_port port;
2058         int error;
2059
2060         error = wdp_port_query_by_number(ofproto->wdp, port_no, &port);
2061         if (!error) {
2062             handle_queue_stats_for_port(&port, queue_id, &cbdata);
2063         } else {
2064             /* XXX deal with wdp_port_query_by_number() errors */
2065         }
2066         wdp_port_free(&port);
2067     } else {
2068         ofpbuf_delete(cbdata.msg);
2069         return ofp_mkerr(OFPET_QUEUE_OP_FAILED, OFPQOFC_BAD_PORT);
2070     }
2071     queue_tx(cbdata.msg, ofconn, ofconn->reply_counter);
2072
2073     return 0;
2074 }
2075
2076 static int
2077 handle_stats_request(struct ofproto *p, struct ofconn *ofconn,
2078                      struct ofp_header *oh)
2079 {
2080     struct ofp_stats_request *osr;
2081     size_t arg_size;
2082     int error;
2083
2084     error = check_ofp_message_array(oh, OFPT_STATS_REQUEST, sizeof *osr,
2085                                     1, &arg_size);
2086     if (error) {
2087         return error;
2088     }
2089     osr = (struct ofp_stats_request *) oh;
2090
2091     switch (ntohs(osr->type)) {
2092     case OFPST_DESC:
2093         return handle_desc_stats_request(p, ofconn, osr);
2094
2095     case OFPST_FLOW:
2096         return handle_flow_stats_request(p, ofconn, osr, arg_size);
2097
2098     case OFPST_AGGREGATE:
2099         return handle_aggregate_stats_request(p, ofconn, osr, arg_size);
2100
2101     case OFPST_TABLE:
2102         return handle_table_stats_request(p, ofconn, osr);
2103
2104     case OFPST_PORT:
2105         return handle_port_stats_request(p, ofconn, osr, arg_size);
2106
2107     case OFPST_QUEUE:
2108         return handle_queue_stats_request(p, ofconn, osr, arg_size);
2109
2110     case OFPST_VENDOR:
2111         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_VENDOR);
2112
2113     default:
2114         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_STAT);
2115     }
2116 }
2117
2118 /* Implements OFPFC_ADD and the cases for OFPFC_MODIFY and OFPFC_MODIFY_STRICT
2119  * in which no matching flow already exists in the flow table.
2120  *
2121  * Adds the flow specified by 'ofm', which is followed by 'n_actions'
2122  * ofp_actions, to 'p''s flow table.  Returns 0 on success or an OpenFlow error
2123  * code as encoded by ofp_mkerr() on failure.
2124  *
2125  * 'ofconn' is used to retrieve the packet buffer specified in ofm->buffer_id,
2126  * if any. */
2127 static int
2128 add_flow(struct ofproto *p, struct ofconn *ofconn,
2129          const struct ofp_flow_mod *ofm, size_t n_actions)
2130 {
2131     struct wdp_rule *rule;
2132     struct wdp_flow_put put;
2133     struct ofpbuf *packet;
2134     uint16_t in_port;
2135     flow_t flow;
2136     int error;
2137
2138     flow_from_match(&ofm->match, ntohs(ofm->priority), p->tun_id_from_cookie,
2139                     ofm->cookie, &flow);
2140     if (ofm->flags & htons(OFPFF_CHECK_OVERLAP)
2141         && wdp_flow_overlaps(p->wdp, &flow)) {
2142         return ofp_mkerr(OFPET_FLOW_MOD_FAILED, OFPFMFC_OVERLAP);
2143     }
2144
2145     put.flags = WDP_PUT_CREATE | WDP_PUT_MODIFY | WDP_PUT_ALL;
2146     put.flow = &flow;
2147     put.actions = (const union ofp_action *) ofm->actions;
2148     put.n_actions = n_actions;
2149     put.idle_timeout = ntohs(ofm->idle_timeout);
2150     put.hard_timeout = ntohs(ofm->hard_timeout);
2151     put.ofp_table_id = flow_mod_table_id(ofconn, ofm);
2152     error = wdp_flow_put(p->wdp, &put, NULL, &rule);
2153     if (error) {
2154         /* XXX wdp_flow_put should return OpenFlow error code. */
2155         return error;
2156     }
2157     ofproto_rule_init(rule);
2158
2159     if (ofm->buffer_id != htonl(UINT32_MAX)) {
2160         error = pktbuf_retrieve(ofconn->pktbuf, ntohl(ofm->buffer_id),
2161                                 &packet, &in_port);
2162         if (!error) {
2163             wdp_flow_inject(p->wdp, rule, in_port, packet);
2164             ofpbuf_delete(packet);
2165         }
2166     }
2167
2168     return 0;
2169 }
2170
2171 static struct wdp_rule *
2172 find_flow_strict(struct ofproto *p, const struct ofconn *ofconn,
2173                  const struct ofp_flow_mod *ofm)
2174 {
2175     uint8_t table_id;
2176     flow_t flow;
2177
2178     flow_from_match(&ofm->match, ntohs(ofm->priority),
2179                     p->tun_id_from_cookie, ofm->cookie, &flow);
2180     table_id = flow_mod_table_id(ofconn, ofm);
2181     return wdp_flow_get(p->wdp, &flow, table_id_to_include(table_id));
2182 }
2183
2184 static int
2185 send_buffered_packet(struct ofproto *ofproto, struct ofconn *ofconn,
2186                      struct wdp_rule *rule, const struct ofp_flow_mod *ofm)
2187 {
2188     struct ofpbuf *packet;
2189     uint16_t in_port;
2190     int error;
2191
2192     if (ofm->buffer_id == htonl(UINT32_MAX)) {
2193         return 0;
2194     }
2195
2196     error = pktbuf_retrieve(ofconn->pktbuf, ntohl(ofm->buffer_id),
2197                             &packet, &in_port);
2198     if (error) {
2199         return error;
2200     }
2201
2202     wdp_flow_inject(ofproto->wdp, rule, in_port, packet);
2203     ofpbuf_delete(packet);
2204
2205     return 0;
2206 }
2207 \f
2208 /* OFPFC_MODIFY and OFPFC_MODIFY_STRICT. */
2209
2210 struct modify_flows_cbdata {
2211     struct ofproto *ofproto;
2212     const struct ofp_flow_mod *ofm;
2213     size_t n_actions;
2214     struct wdp_rule *match;
2215 };
2216
2217 static int modify_flow(struct ofproto *, const struct ofp_flow_mod *,
2218                        size_t n_actions, struct wdp_rule *);
2219 static int modify_flows_cb(struct wdp_rule *, void *cbdata_);
2220
2221 /* Implements OFPFC_ADD and OFPFC_MODIFY.  Returns 0 on success or an OpenFlow
2222  * error code as encoded by ofp_mkerr() on failure.
2223  *
2224  * 'ofconn' is used to retrieve the packet buffer specified in ofm->buffer_id,
2225  * if any. */
2226 static int
2227 modify_flows_loose(struct ofproto *p, struct ofconn *ofconn,
2228                    const struct ofp_flow_mod *ofm, size_t n_actions)
2229 {
2230     struct modify_flows_cbdata cbdata;
2231     uint8_t table_id;
2232     flow_t target;
2233
2234     cbdata.ofproto = p;
2235     cbdata.ofm = ofm;
2236     cbdata.n_actions = n_actions;
2237     cbdata.match = NULL;
2238
2239     flow_from_match(&ofm->match, 0, p->tun_id_from_cookie, ofm->cookie,
2240                     &target);
2241
2242     table_id = flow_mod_table_id(ofconn, ofm);
2243     wdp_flow_for_each_match(p->wdp, &target, table_id_to_include(table_id),
2244                             modify_flows_cb, &cbdata);
2245     if (cbdata.match) {
2246         /* This credits the packet to whichever flow happened to match last.
2247          * That's weird.  Maybe we should do a lookup for the flow that
2248          * actually matches the packet?  Who knows. */
2249         send_buffered_packet(p, ofconn, cbdata.match, ofm);
2250         return 0;
2251     } else {
2252         return add_flow(p, ofconn, ofm, n_actions);
2253     }
2254 }
2255
2256 /* Implements OFPFC_MODIFY_STRICT.  Returns 0 on success or an OpenFlow error
2257  * code as encoded by ofp_mkerr() on failure.
2258  *
2259  * 'ofconn' is used to retrieve the packet buffer specified in ofm->buffer_id,
2260  * if any. */
2261 static int
2262 modify_flow_strict(struct ofproto *p, struct ofconn *ofconn,
2263                    struct ofp_flow_mod *ofm, size_t n_actions)
2264 {
2265     struct wdp_rule *rule = find_flow_strict(p, ofconn, ofm);
2266     if (rule && !rule_is_hidden(rule)) {
2267         modify_flow(p, ofm, n_actions, rule);
2268         return send_buffered_packet(p, ofconn, rule, ofm);
2269     } else {
2270         return add_flow(p, ofconn, ofm, n_actions);
2271     }
2272 }
2273
2274 /* Callback for modify_flows_loose(). */
2275 static int
2276 modify_flows_cb(struct wdp_rule *rule, void *cbdata_)
2277 {
2278     struct modify_flows_cbdata *cbdata = cbdata_;
2279
2280     if (!rule_is_hidden(rule)) {
2281         cbdata->match = rule;
2282         modify_flow(cbdata->ofproto, cbdata->ofm, cbdata->n_actions, rule);
2283     }
2284     return 0;
2285 }
2286
2287 /* Implements core of OFPFC_MODIFY and OFPFC_MODIFY_STRICT where 'rule' has
2288  * been identified as a flow in 'p''s flow table to be modified, by changing
2289  * the rule's actions to match those in 'ofm' (which is followed by 'n_actions'
2290  * ofp_action[] structures). */
2291 static int
2292 modify_flow(struct ofproto *p, const struct ofp_flow_mod *ofm,
2293             size_t n_actions, struct wdp_rule *rule)
2294 {
2295     const struct ofp_action_header *actions = ofm->actions;
2296     struct ofproto_rule *ofproto_rule = ofproto_rule_cast(rule);
2297     struct wdp_flow_put put;
2298
2299     ofproto_rule->flow_cookie = ofm->cookie;
2300
2301     /* If the actions are the same, do nothing. */
2302     if (n_actions == rule->n_actions
2303         && !memcmp(ofm->actions, rule->actions, sizeof *actions * n_actions))
2304     {
2305         return 0;
2306     }
2307
2308     put.flags = WDP_PUT_MODIFY | WDP_PUT_ACTIONS;
2309     put.flow = &rule->cr.flow;
2310     put.actions = (const union ofp_action *) actions;
2311     put.n_actions = n_actions;
2312     put.idle_timeout = put.hard_timeout = 0;
2313     put.ofp_table_id = rule->ofp_table_id;
2314     return wdp_flow_put(p->wdp, &put, NULL, NULL);
2315 }
2316 \f
2317 /* OFPFC_DELETE implementation. */
2318
2319 struct delete_flows_cbdata {
2320     struct ofproto *ofproto;
2321     uint16_t out_port;
2322 };
2323
2324 static int delete_flows_cb(struct wdp_rule *, void *cbdata_);
2325 static void delete_flow_core(struct ofproto *, struct wdp_rule *,
2326                              uint16_t out_port);
2327
2328 /* Implements OFPFC_DELETE. */
2329 static void
2330 delete_flows_loose(struct ofproto *p, const struct ofconn *ofconn,
2331                    const struct ofp_flow_mod *ofm)
2332 {
2333     struct delete_flows_cbdata cbdata;
2334     uint8_t table_id;
2335     flow_t target;
2336
2337     cbdata.ofproto = p;
2338     cbdata.out_port = ofm->out_port;
2339
2340     flow_from_match(&ofm->match, 0, p->tun_id_from_cookie, ofm->cookie,
2341                     &target);
2342     table_id = flow_mod_table_id(ofconn, ofm);
2343
2344     wdp_flow_for_each_match(p->wdp, &target, table_id_to_include(table_id),
2345                             delete_flows_cb, &cbdata);
2346 }
2347
2348 /* Implements OFPFC_DELETE_STRICT. */
2349 static void
2350 delete_flow_strict(struct ofproto *p, const struct ofconn *ofconn,
2351                    struct ofp_flow_mod *ofm)
2352 {
2353     struct wdp_rule *rule = find_flow_strict(p, ofconn, ofm);
2354     if (rule) {
2355         delete_flow_core(p, rule, ofm->out_port);
2356     }
2357 }
2358
2359 /* Callback for delete_flows_loose(). */
2360 static int
2361 delete_flows_cb(struct wdp_rule *rule, void *cbdata_)
2362 {
2363     struct delete_flows_cbdata *cbdata = cbdata_;
2364
2365     delete_flow_core(cbdata->ofproto, rule, cbdata->out_port);
2366     return 0;
2367 }
2368
2369 /* Implements core of OFPFC_DELETE and OFPFC_DELETE_STRICT where 'rule' has
2370  * been identified as a flow to delete from 'p''s flow table, by deleting the
2371  * flow and sending out a OFPT_FLOW_REMOVED message to any interested
2372  * controller.
2373  *
2374  * Will not delete 'rule' if it is hidden.  Will delete 'rule' only if
2375  * 'out_port' is htons(OFPP_NONE) or if 'rule' actually outputs to the
2376  * specified 'out_port'. */
2377 static void
2378 delete_flow_core(struct ofproto *p, struct wdp_rule *rule, uint16_t out_port)
2379 {
2380     if (rule_is_hidden(rule)) {
2381         return;
2382     }
2383
2384     if (out_port != htons(OFPP_NONE) && !rule_has_out_port(rule, out_port)) {
2385         return;
2386     }
2387
2388     delete_flow(p, rule, OFPRR_DELETE);
2389 }
2390 \f
2391 static int
2392 handle_flow_mod(struct ofproto *p, struct ofconn *ofconn,
2393                 struct ofp_flow_mod *ofm)
2394 {
2395     struct ofp_match orig_match;
2396     size_t n_actions;
2397     int error;
2398
2399     error = reject_slave_controller(ofconn, &ofm->header);
2400     if (error) {
2401         return error;
2402     }
2403     error = check_ofp_message_array(&ofm->header, OFPT_FLOW_MOD, sizeof *ofm,
2404                                     sizeof *ofm->actions, &n_actions);
2405     if (error) {
2406         return error;
2407     }
2408
2409     /* We do not support the emergency flow cache.  It will hopefully
2410      * get dropped from OpenFlow in the near future. */
2411     if (ofm->flags & htons(OFPFF_EMERG)) {
2412         /* There isn't a good fit for an error code, so just state that the
2413          * flow table is full. */
2414         return ofp_mkerr(OFPET_FLOW_MOD_FAILED, OFPFMFC_ALL_TABLES_FULL);
2415     }
2416
2417     /* Normalize ofp->match.  If normalization actually changes anything, then
2418      * log the differences. */
2419     ofm->match.pad1[0] = ofm->match.pad2[0] = 0;
2420     orig_match = ofm->match;
2421     normalize_match(&ofm->match);
2422     if (memcmp(&ofm->match, &orig_match, sizeof orig_match)) {
2423         static struct vlog_rate_limit normal_rl = VLOG_RATE_LIMIT_INIT(1, 1);
2424         if (!VLOG_DROP_INFO(&normal_rl)) {
2425             char *old = ofp_match_to_literal_string(&orig_match);
2426             char *new = ofp_match_to_literal_string(&ofm->match);
2427             VLOG_INFO("%s: normalization changed ofp_match, details:",
2428                       rconn_get_name(ofconn->rconn));
2429             VLOG_INFO(" pre: %s", old);
2430             VLOG_INFO("post: %s", new);
2431             free(old);
2432             free(new);
2433         }
2434     }
2435
2436     if (!ofm->match.wildcards) {
2437         ofm->priority = htons(UINT16_MAX);
2438     }
2439
2440     error = validate_actions((const union ofp_action *) ofm->actions,
2441                              n_actions, p->max_ports);
2442     if (error) {
2443         return error;
2444     }
2445
2446     if (!ofconn->flow_mod_table_id && ofm->command & htons(0xff00)) {
2447         static struct vlog_rate_limit table_id_rl = VLOG_RATE_LIMIT_INIT(1, 1);
2448         VLOG_WARN_RL(&table_id_rl, "%s: flow_mod table_id feature must be "
2449                      "enabled with NXT_FLOW_MOD_TABLE_ID",
2450                      rconn_get_name(ofconn->rconn));
2451         return ofp_mkerr(OFPET_FLOW_MOD_FAILED, OFPFMFC_BAD_COMMAND);
2452     }
2453
2454     switch (ntohs(ofm->command) & 0xff) {
2455     case OFPFC_ADD:
2456         return modify_flows_loose(p, ofconn, ofm, n_actions);
2457
2458     case OFPFC_MODIFY:
2459         return modify_flow_strict(p, ofconn, ofm, n_actions);
2460
2461     case OFPFC_MODIFY_STRICT:
2462         return modify_flow_strict(p, ofconn, ofm, n_actions);
2463
2464     case OFPFC_DELETE:
2465         delete_flows_loose(p, ofconn, ofm);
2466         return 0;
2467
2468     case OFPFC_DELETE_STRICT:
2469         delete_flow_strict(p, ofconn, ofm);
2470         return 0;
2471
2472     default:
2473         return ofp_mkerr(OFPET_FLOW_MOD_FAILED, OFPFMFC_BAD_COMMAND);
2474     }
2475 }
2476
2477 static int
2478 handle_tun_id_from_cookie(struct ofproto *p, struct nxt_tun_id_cookie *msg)
2479 {
2480     int error;
2481
2482     error = check_ofp_message(&msg->header, OFPT_VENDOR, sizeof *msg);
2483     if (error) {
2484         return error;
2485     }
2486
2487     p->tun_id_from_cookie = !!msg->set;
2488     return 0;
2489 }
2490
2491 static int
2492 handle_flow_mod_table_id(struct ofconn *ofconn,
2493                          struct nxt_flow_mod_table_id *msg)
2494 {
2495     int error;
2496
2497     error = check_ofp_message(&msg->header, OFPT_VENDOR, sizeof *msg);
2498     if (error) {
2499         return error;
2500     }
2501
2502     ofconn->flow_mod_table_id = !!msg->set;
2503     return 0;
2504 }
2505
2506 static int
2507 handle_role_request(struct ofproto *ofproto,
2508                     struct ofconn *ofconn, struct nicira_header *msg)
2509 {
2510     struct nx_role_request *nrr;
2511     struct nx_role_request *reply;
2512     struct ofpbuf *buf;
2513     uint32_t role;
2514
2515     if (ntohs(msg->header.length) != sizeof *nrr) {
2516         VLOG_WARN_RL(&rl, "received role request of length %u (expected %zu)",
2517                      ntohs(msg->header.length), sizeof *nrr);
2518         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
2519     }
2520     nrr = (struct nx_role_request *) msg;
2521
2522     if (ofconn->type != OFCONN_PRIMARY) {
2523         VLOG_WARN_RL(&rl, "ignoring role request on non-controller "
2524                      "connection");
2525         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_EPERM);
2526     }
2527
2528     role = ntohl(nrr->role);
2529     if (role != NX_ROLE_OTHER && role != NX_ROLE_MASTER
2530         && role != NX_ROLE_SLAVE) {
2531         VLOG_WARN_RL(&rl, "received request for unknown role %"PRIu32, role);
2532
2533         /* There's no good error code for this. */
2534         return ofp_mkerr(OFPET_BAD_REQUEST, -1);
2535     }
2536
2537     if (role == NX_ROLE_MASTER) {
2538         struct ofconn *other;
2539
2540         HMAP_FOR_EACH (other, struct ofconn, hmap_node,
2541                        &ofproto->controllers) {
2542             if (other->role == NX_ROLE_MASTER) {
2543                 other->role = NX_ROLE_SLAVE;
2544             }
2545         }
2546     }
2547     ofconn->role = role;
2548
2549     reply = make_openflow_xid(sizeof *reply, OFPT_VENDOR, msg->header.xid,
2550                               &buf);
2551     reply->nxh.vendor = htonl(NX_VENDOR_ID);
2552     reply->nxh.subtype = htonl(NXT_ROLE_REPLY);
2553     reply->role = htonl(role);
2554     queue_tx(buf, ofconn, ofconn->reply_counter);
2555
2556     return 0;
2557 }
2558
2559 static int
2560 handle_vendor(struct ofproto *p, struct ofconn *ofconn, void *msg)
2561 {
2562     struct ofp_vendor_header *ovh = msg;
2563     struct nicira_header *nh;
2564
2565     if (ntohs(ovh->header.length) < sizeof(struct ofp_vendor_header)) {
2566         VLOG_WARN_RL(&rl, "received vendor message of length %u "
2567                           "(expected at least %zu)",
2568                    ntohs(ovh->header.length), sizeof(struct ofp_vendor_header));
2569         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
2570     }
2571     if (ovh->vendor != htonl(NX_VENDOR_ID)) {
2572         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_VENDOR);
2573     }
2574     if (ntohs(ovh->header.length) < sizeof(struct nicira_header)) {
2575         VLOG_WARN_RL(&rl, "received Nicira vendor message of length %u "
2576                           "(expected at least %zu)",
2577                      ntohs(ovh->header.length), sizeof(struct nicira_header));
2578         return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_LEN);
2579     }
2580
2581     nh = msg;
2582     switch (ntohl(nh->subtype)) {
2583     case NXT_STATUS_REQUEST:
2584         return switch_status_handle_request(p->switch_status, ofconn->rconn,
2585                                             msg);
2586
2587     case NXT_TUN_ID_FROM_COOKIE:
2588         return handle_tun_id_from_cookie(p, msg);
2589
2590     case NXT_FLOW_MOD_TABLE_ID:
2591         return handle_flow_mod_table_id(ofconn, msg);
2592
2593     case NXT_ROLE_REQUEST:
2594         return handle_role_request(p, ofconn, msg);
2595     }
2596
2597     return ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_SUBTYPE);
2598 }
2599
2600 static int
2601 handle_barrier_request(struct ofconn *ofconn, struct ofp_header *oh)
2602 {
2603     struct ofp_header *ob;
2604     struct ofpbuf *buf;
2605
2606     /* Currently, everything executes synchronously, so we can just
2607      * immediately send the barrier reply. */
2608     ob = make_openflow_xid(sizeof *ob, OFPT_BARRIER_REPLY, oh->xid, &buf);
2609     queue_tx(buf, ofconn, ofconn->reply_counter);
2610     return 0;
2611 }
2612
2613 static void
2614 handle_openflow(struct ofconn *ofconn, struct ofproto *p,
2615                 struct ofpbuf *ofp_msg)
2616 {
2617     struct ofp_header *oh = ofp_msg->data;
2618     int error;
2619
2620     COVERAGE_INC(ofproto_recv_openflow);
2621     switch (oh->type) {
2622     case OFPT_ECHO_REQUEST:
2623         error = handle_echo_request(ofconn, oh);
2624         break;
2625
2626     case OFPT_ECHO_REPLY:
2627         error = 0;
2628         break;
2629
2630     case OFPT_FEATURES_REQUEST:
2631         error = handle_features_request(p, ofconn, oh);
2632         break;
2633
2634     case OFPT_GET_CONFIG_REQUEST:
2635         error = handle_get_config_request(p, ofconn, oh);
2636         break;
2637
2638     case OFPT_SET_CONFIG:
2639         error = handle_set_config(p, ofconn, ofp_msg->data);
2640         break;
2641
2642     case OFPT_PACKET_OUT:
2643         error = handle_packet_out(p, ofconn, ofp_msg->data);
2644         break;
2645
2646     case OFPT_PORT_MOD:
2647         error = handle_port_mod(p, ofconn, oh);
2648         break;
2649
2650     case OFPT_FLOW_MOD:
2651         error = handle_flow_mod(p, ofconn, ofp_msg->data);
2652         break;
2653
2654     case OFPT_STATS_REQUEST:
2655         error = handle_stats_request(p, ofconn, oh);
2656         break;
2657
2658     case OFPT_VENDOR:
2659         error = handle_vendor(p, ofconn, ofp_msg->data);
2660         break;
2661
2662     case OFPT_BARRIER_REQUEST:
2663         error = handle_barrier_request(ofconn, oh);
2664         break;
2665
2666     default:
2667         if (VLOG_IS_WARN_ENABLED()) {
2668             char *s = ofp_to_string(oh, ntohs(oh->length), 2);
2669             VLOG_DBG_RL(&rl, "OpenFlow message ignored: %s", s);
2670             free(s);
2671         }
2672         error = ofp_mkerr(OFPET_BAD_REQUEST, OFPBRC_BAD_TYPE);
2673         break;
2674     }
2675
2676     if (error) {
2677         send_error_oh(ofconn, ofp_msg->data, error);
2678     }
2679 }
2680 \f
2681 static void
2682 handle_flow_miss(struct ofproto *p, struct wdp_packet *packet)
2683 {
2684     struct wdp_rule *rule;
2685     flow_t flow;
2686
2687     flow_extract(packet->payload, packet->tun_id, packet->in_port, &flow);
2688     rule = wdp_flow_match(p->wdp, &flow);
2689     if (!rule) {
2690         /* Don't send a packet-in if OFPPC_NO_PACKET_IN asserted. */
2691         struct wdp_port port;
2692
2693         if (!wdp_port_query_by_number(p->wdp, packet->in_port, &port)) {
2694             bool no_packet_in = (port.opp.config & OFPPC_NO_PACKET_IN) != 0;
2695             wdp_port_free(&port);
2696             if (no_packet_in) {
2697                 COVERAGE_INC(ofproto_no_packet_in);
2698                 wdp_packet_destroy(packet);
2699                 return;
2700             }
2701         } else {
2702             VLOG_WARN_RL(&rl, "packet-in on unknown port %"PRIu16,
2703                          packet->in_port);
2704         }
2705
2706         COVERAGE_INC(ofproto_packet_in);
2707         send_packet_in(p, packet);
2708         return;
2709     }
2710
2711     wdp_flow_inject(p->wdp, rule, packet->in_port, packet->payload);
2712
2713     if (rule->cr.flow.priority == FAIL_OPEN_PRIORITY) {
2714         /*
2715          * Extra-special case for fail-open mode.
2716          *
2717          * We are in fail-open mode and the packet matched the fail-open rule,
2718          * but we are connected to a controller too.  We should send the packet
2719          * up to the controller in the hope that it will try to set up a flow
2720          * and thereby allow us to exit fail-open.
2721          *
2722          * See the top-level comment in fail-open.c for more information.
2723          */
2724         send_packet_in(p, packet);
2725     } else {
2726         wdp_packet_destroy(packet);
2727     }
2728 }
2729
2730 static void
2731 handle_wdp_packet(struct ofproto *p, struct wdp_packet *packet)
2732 {
2733     switch (packet->channel) {
2734     case WDP_CHAN_ACTION:
2735         COVERAGE_INC(ofproto_ctlr_action);
2736         send_packet_in(p, packet);
2737         break;
2738
2739     case WDP_CHAN_SFLOW:
2740         /* XXX */
2741         wdp_packet_destroy(packet);
2742         break;
2743
2744     case WDP_CHAN_MISS:
2745         handle_flow_miss(p, packet);
2746         break;
2747
2748     case WDP_N_CHANS:
2749     default:
2750         wdp_packet_destroy(packet);
2751         VLOG_WARN_RL(&rl, "received message on unexpected channel %d",
2752                      (int) packet->channel);
2753         break;
2754     }
2755 }
2756 \f
2757 static struct ofpbuf *
2758 compose_flow_removed(struct ofproto *p, const struct wdp_rule *rule,
2759                      uint8_t reason)
2760 {
2761     long long int tdiff = time_msec() - rule->created;
2762     uint32_t sec = tdiff / 1000;
2763     uint32_t msec = tdiff - (sec * 1000);
2764     struct ofp_flow_removed *ofr;
2765     struct ofpbuf *buf;
2766
2767     ofr = make_openflow(sizeof *ofr, OFPT_FLOW_REMOVED, &buf);
2768     flow_to_match(&rule->cr.flow, p->tun_id_from_cookie, &ofr->match);
2769     ofr->cookie = ofproto_rule_cast(rule)->flow_cookie;
2770     ofr->priority = htons(rule->cr.flow.priority);
2771     ofr->reason = reason;
2772     ofr->duration_sec = htonl(sec);
2773     ofr->duration_nsec = htonl(msec * 1000000);
2774     ofr->idle_timeout = htons(rule->idle_timeout);
2775
2776     return buf;
2777 }
2778
2779 static void
2780 delete_flow(struct ofproto *p, struct wdp_rule *rule, uint8_t reason)
2781 {
2782     /* We limit the maximum number of queued flow expirations it by accounting
2783      * them under the counter for replies.  That works because preventing
2784      * OpenFlow requests from being processed also prevents new flows from
2785      * being added (and expiring).  (It also prevents processing OpenFlow
2786      * requests that would not add new flows, so it is imperfect.) */
2787
2788     struct ofproto_rule *ofproto_rule = ofproto_rule_cast(rule);
2789     struct wdp_flow_stats stats;
2790     struct ofpbuf *buf;
2791
2792     if (ofproto_rule->send_flow_removed) {
2793         /* Compose most of the ofp_flow_removed before 'rule' is destroyed. */
2794         buf = compose_flow_removed(p, rule, reason);
2795     } else {
2796         buf = NULL;
2797     }
2798
2799     if (wdp_flow_delete(p->wdp, rule, &stats)) {
2800         return;
2801     }
2802
2803     if (buf) {
2804         struct ofp_flow_removed *ofr;
2805         struct ofconn *prev = NULL;
2806         struct ofconn *ofconn;
2807
2808         /* Compose the parts of the ofp_flow_removed that require stats. */
2809         ofr = buf->data;
2810         ofr->packet_count = htonll(stats.n_packets);
2811         ofr->byte_count = htonll(stats.n_bytes);
2812
2813         LIST_FOR_EACH (ofconn, struct ofconn, node, &p->all_conns) {
2814             if (rconn_is_connected(ofconn->rconn)) {
2815                 if (prev) {
2816                     queue_tx(ofpbuf_clone(buf), prev, prev->reply_counter);
2817                 }
2818                 prev = ofconn;
2819             }
2820         }
2821         if (prev) {
2822             queue_tx(buf, prev, prev->reply_counter);
2823         } else {
2824             ofpbuf_delete(buf);
2825         }
2826     }
2827     free(ofproto_rule);
2828 }
2829
2830 /* pinsched callback for sending 'packet' on 'ofconn'. */
2831 static void
2832 do_send_packet_in(struct wdp_packet *packet, void *ofconn_)
2833 {
2834     struct ofconn *ofconn = ofconn_;
2835
2836     rconn_send_with_limit(ofconn->rconn, packet->payload,
2837                           ofconn->packet_in_counter, 100);
2838     packet->payload = NULL;
2839     wdp_packet_destroy(packet);
2840 }
2841
2842 /* Takes 'packet', which has been converted with do_convert_to_packet_in(), and
2843  * finalizes its content for sending on 'ofconn', and passes it to 'ofconn''s
2844  * packet scheduler for sending.
2845  *
2846  * 'max_len' specifies the maximum number of bytes of the packet to send on
2847  * 'ofconn' (INT_MAX specifies no limit).
2848  *
2849  * If 'clone' is true, the caller retains ownership of 'packet'.  Otherwise,
2850  * ownership is transferred to this function. */
2851 static void
2852 schedule_packet_in(struct ofconn *ofconn, struct wdp_packet *packet,
2853                    int max_len, bool clone)
2854 {
2855     struct ofproto *ofproto = ofconn->ofproto;
2856     struct ofp_packet_in *opi = packet->payload->data;
2857     int send_len, trim_size;
2858     uint32_t buffer_id;
2859
2860     /* Get buffer. */
2861     if (opi->reason == OFPR_ACTION) {
2862         buffer_id = UINT32_MAX;
2863     } else if (ofproto->fail_open && fail_open_is_active(ofproto->fail_open)) {
2864         buffer_id = pktbuf_get_null();
2865     } else if (!ofconn->pktbuf) {
2866         buffer_id = UINT32_MAX;
2867     } else {
2868         struct ofpbuf payload;
2869         payload.data = opi->data;
2870         payload.size = (packet->payload->size
2871                         - offsetof(struct ofp_packet_in, data));
2872         buffer_id = pktbuf_save(ofconn->pktbuf, &payload, packet->in_port);
2873     }
2874
2875     /* Figure out how much of the packet to send. */
2876     send_len = ntohs(opi->total_len);
2877     if (buffer_id != UINT32_MAX) {
2878         send_len = MIN(send_len, ofconn->miss_send_len);
2879     }
2880     send_len = MIN(send_len, max_len);
2881
2882     /* Adjust packet length and clone if necessary. */
2883     trim_size = offsetof(struct ofp_packet_in, data) + send_len;
2884     if (clone) {
2885         packet = wdp_packet_clone(packet, trim_size);
2886         opi = packet->payload->data;
2887     } else {
2888         packet->payload->size = trim_size;
2889     }
2890
2891     /* Update packet headers. */
2892     opi->buffer_id = htonl(buffer_id);
2893     update_openflow_length(packet->payload);
2894
2895     /* Hand over to packet scheduler.  It might immediately call into
2896      * do_send_packet_in() or it might buffer it for a while (until a later
2897      * call to pinsched_run()). */
2898     pinsched_send(ofconn->schedulers[opi->reason], packet->in_port,
2899                   packet, do_send_packet_in, ofconn);
2900 }
2901
2902 /* Converts 'packet->payload' to a struct ofp_packet_in.  It must have
2903  * sufficient headroom to do so (e.g. as returned by xfif_recv()).
2904  *
2905  * The conversion is not complete: the caller still needs to trim any unneeded
2906  * payload off the end of the buffer, set the length in the OpenFlow header,
2907  * and set buffer_id.  Those require us to know the controller settings and so
2908  * must be done on a per-controller basis.
2909  *
2910  * Returns the maximum number of bytes of the packet that should be sent to
2911  * the controller (INT_MAX if no limit). */
2912 static int
2913 do_convert_to_packet_in(struct wdp_packet *packet)
2914 {
2915     uint16_t total_len = packet->payload->size;
2916     struct ofp_packet_in *opi;
2917
2918     /* Repurpose packet buffer by overwriting header. */
2919     opi = ofpbuf_push_zeros(packet->payload,
2920                             offsetof(struct ofp_packet_in, data));
2921     opi->header.version = OFP_VERSION;
2922     opi->header.type = OFPT_PACKET_IN;
2923     opi->total_len = htons(total_len);
2924     opi->in_port = htons(packet->in_port);
2925     if (packet->channel == WDP_CHAN_MISS) {
2926         opi->reason = OFPR_NO_MATCH;
2927         return INT_MAX;
2928     } else {
2929         opi->reason = OFPR_ACTION;
2930         return packet->send_len;
2931     }
2932 }
2933
2934 /* Given 'packet' with channel WDP_CHAN_ACTION or WDP_CHAN_MISS, sends an
2935  * OFPT_PACKET_IN message to each OpenFlow controller as necessary according to
2936  * their individual configurations.
2937  *
2938  * 'packet->payload' must have sufficient headroom to convert it into a struct
2939  * ofp_packet_in (e.g. as returned by dpif_recv()).
2940  *
2941  * Takes ownership of 'packet'. */
2942 static void
2943 send_packet_in(struct ofproto *ofproto, struct wdp_packet *packet)
2944 {
2945     struct ofconn *ofconn, *prev;
2946     int max_len;
2947
2948     max_len = do_convert_to_packet_in(packet);
2949
2950     prev = NULL;
2951     LIST_FOR_EACH (ofconn, struct ofconn, node, &ofproto->all_conns) {
2952         if (ofconn_receives_async_msgs(ofconn)) {
2953             if (prev) {
2954                 schedule_packet_in(prev, packet, max_len, true);
2955             }
2956             prev = ofconn;
2957         }
2958     }
2959     if (prev) {
2960         schedule_packet_in(prev, packet, max_len, false);
2961     } else {
2962         wdp_packet_destroy(packet);
2963     }
2964 }
2965
2966 static uint64_t
2967 pick_datapath_id(const struct ofproto *ofproto)
2968 {
2969     struct wdp_port port;
2970
2971     if (!wdp_port_query_by_number(ofproto->wdp, OFPP_LOCAL, &port)) {
2972         uint8_t ea[ETH_ADDR_LEN];
2973         int error;
2974
2975         error = netdev_get_etheraddr(port.netdev, ea);
2976         if (!error) {
2977             wdp_port_free(&port);
2978             return eth_addr_to_uint64(ea);
2979         }
2980         VLOG_WARN("could not get MAC address for %s (%s)",
2981                   netdev_get_name(port.netdev), strerror(error));
2982         wdp_port_free(&port);
2983     }
2984
2985     return ofproto->fallback_dpid;
2986 }
2987
2988 static uint64_t
2989 pick_fallback_dpid(void)
2990 {
2991     uint8_t ea[ETH_ADDR_LEN];
2992     eth_addr_nicira_random(ea);
2993     return eth_addr_to_uint64(ea);
2994 }