From 5733b229177144b9396a1920c9d5ab61430ba654 Mon Sep 17 00:00:00 2001 From: Yiannis Yiakoumis Date: Tue, 1 Dec 2009 18:52:43 -0800 Subject: [PATCH] Updated datapath for slicing feature This commit updates the userspace datapath to handle the slicing mechanism. -> Keep list of queues per port. -> Map flows to queues -> Report queue configuration and statistics -> Send appropriate error messages --- lib/dhcp-client.c | 2 +- udatapath/datapath.c | 306 ++++++++++++++++++++++++++++++++++++------- udatapath/datapath.h | 21 ++- udatapath/dp_act.c | 37 +++++- 4 files changed, 310 insertions(+), 56 deletions(-) diff --git a/lib/dhcp-client.c b/lib/dhcp-client.c index 5cbd297..3f4b9eb 100644 --- a/lib/dhcp-client.c +++ b/lib/dhcp-client.c @@ -1048,7 +1048,7 @@ do_send_msg(struct dhclient *cli, const struct dhcp_msg *msg) } else { VLOG_INFO("sending %s", dhcp_type_name(msg->type)); } - error = netdev_send(cli->netdev, &b); + error = netdev_send(cli->netdev, &b, 0); if (error) { VLOG_ERR("send failed on %s: %s", netdev_get_name(cli->netdev), strerror(error)); diff --git a/udatapath/datapath.c b/udatapath/datapath.c index 178f2d9..8a6491f 100644 --- a/udatapath/datapath.c +++ b/udatapath/datapath.c @@ -1,6 +1,7 @@ +//234567890123456789012345678901234567890123456789012345678901234567890123456789 /* Copyright (c) 2008, 2009 The Board of Trustees of The Leland Stanford * Junior University - * + * * We are making the OpenFlow specification and associated documentation * (Software) available for public use and benefit with the expectation * that others will use, modify and enhance the Software and contribute @@ -13,10 +14,10 @@ * distribute, sublicense, and/or sell copies of the Software, and to * permit persons to whom the Software is furnished to do so, subject to * the following conditions: - * + * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. - * + * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND @@ -25,7 +26,7 @@ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE * SOFTWARE. - * + * * The name and trademarks of copyright holder(s) may NOT be used in * advertising or publicity pertaining to the Software or any * derivatives without specific, written prior permission. @@ -41,11 +42,11 @@ #include "chain.h" #include "csum.h" #include "flow.h" -#include "netdev.h" #include "ofpbuf.h" #include "openflow/openflow.h" #include "openflow/nicira-ext.h" #include "openflow/private-ext.h" +#include "openflow/openflow-ext.h" #include "packets.h" #include "poll-loop.h" #include "rconn.h" @@ -55,6 +56,7 @@ #include "vconn.h" #include "xtoxll.h" #include "private-msg.h" +#include "of_ext_msg.h" #include "dp_act.h" #define THIS_MODULE VLM_datapath @@ -66,10 +68,11 @@ extern char sw_desc; extern char serial_num; /* Capabilities supported by this implementation. */ -#define OFP_SUPPORTED_CAPABILITIES ( OFPC_FLOW_STATS \ - | OFPC_TABLE_STATS \ - | OFPC_PORT_STATS \ - | OFPC_MULTI_PHY_TX ) +#define OFP_SUPPORTED_CAPABILITIES ( OFPC_FLOW_STATS \ + | OFPC_TABLE_STATS \ + | OFPC_PORT_STATS \ + | OFPC_QUEUE_STATS \ + | OFPC_MULTI_PHY_TX ) /* Actions supported by this implementation. */ #define OFP_SUPPORTED_ACTIONS ( (1 << OFPAT_OUTPUT) \ @@ -81,7 +84,8 @@ extern char serial_num; | (1 << OFPAT_SET_NW_SRC) \ | (1 << OFPAT_SET_NW_DST) \ | (1 << OFPAT_SET_TP_SRC) \ - | (1 << OFPAT_SET_TP_DST) ) + | (1 << OFPAT_SET_TP_DST) \ + | (1 << OFPAT_ENQUEUE)) /* The origin of a received OpenFlow message, to enable sending a reply. */ struct sender { @@ -137,14 +141,27 @@ uint32_t save_buffer(struct ofpbuf *); static struct ofpbuf *retrieve_buffer(uint32_t id); static void discard_buffer(uint32_t id); -static struct sw_port * -lookup_port(struct datapath *dp, uint16_t port_no) +struct sw_port * +dp_lookup_port(struct datapath *dp, uint16_t port_no) { return (port_no < DP_MAX_PORTS ? &dp->ports[port_no] : port_no == OFPP_LOCAL ? dp->local_port : NULL); } +struct sw_queue * +dp_lookup_queue(struct sw_port *p, uint32_t queue_id) +{ + struct sw_queue *q; + + LIST_FOR_EACH(q, struct sw_queue, node, &p->queue_list) { + if (q->queue_id == queue_id) { + return q; + } + } + return NULL; +} + /* Generates and returns a random datapath id. */ static uint64_t gen_datapath_id(void) @@ -228,8 +245,16 @@ new_port(struct datapath *dp, struct sw_port *port, uint16_t port_no, netdev_name, in6_name); } + error = netdev_setup_slicing(netdev); + if (error) { + VLOG_ERR("failed to configure slicing on %s device",netdev_name); + netdev_close(netdev); + return error; + } + memset(port, '\0', sizeof *port); + list_init(&port->queue_list); port->dp = dp; port->netdev = netdev; port->port_no = port_no; @@ -255,7 +280,7 @@ dp_add_port(struct datapath *dp, const char *netdev) } int -dp_add_local_port(struct datapath *dp, const char *netdev) +dp_add_local_port(struct datapath *dp, const char *netdev) { if (!dp->local_port) { uint8_t ea[ETH_ADDR_LEN]; @@ -306,7 +331,7 @@ dp_run(struct datapath *dp) dp->last_timeout = now; } poll_timer_wait(1000); - + LIST_FOR_EACH_SAFE (p, pn, struct sw_port, node, &dp->port_list) { int error; @@ -382,7 +407,7 @@ remote_run(struct datapath *dp, struct remote *r) } else { VLOG_WARN_RL(&rl, "received too-short OpenFlow message"); } - ofpbuf_delete(buffer); + ofpbuf_delete(buffer); } else { if (r->n_txq < TXQ_LIMIT) { int error = r->cb_dump(dp, r->cb_aux); @@ -406,7 +431,7 @@ remote_run(struct datapath *dp, struct remote *r) } static void -remote_wait(struct remote *r) +remote_wait(struct remote *r) { rconn_run_wait(r->rconn); rconn_recv_wait(r->rconn); @@ -426,7 +451,7 @@ remote_destroy(struct remote *r) } static struct remote * -remote_create(struct datapath *dp, struct rconn *rconn) +remote_create(struct datapath *dp, struct rconn *rconn) { struct remote *remote = xmalloc(sizeof *remote); list_push_back(&dp->remotes, &remote->node); @@ -454,7 +479,7 @@ static void remote_start_dump(struct remote *remote, int (*dump)(struct datapath *, void *), void (*done)(void *), - void *aux) + void *aux) { assert(!remote->cb_dump); remote->cb_dump = dump; @@ -463,7 +488,7 @@ remote_start_dump(struct remote *remote, } void -dp_wait(struct datapath *dp) +dp_wait(struct datapath *dp) { struct sw_port *p; struct remote *r; @@ -499,12 +524,12 @@ output_all(struct datapath *dp, struct ofpbuf *buffer, int in_port, int flood) } if (prev_port != -1) { dp_output_port(dp, ofpbuf_clone(buffer), in_port, prev_port, - false); + 0,false); } prev_port = p->port_no; } if (prev_port != -1) - dp_output_port(dp, buffer, in_port, prev_port, false); + dp_output_port(dp, buffer, in_port, prev_port, 0, false); else ofpbuf_delete(buffer); @@ -512,14 +537,39 @@ output_all(struct datapath *dp, struct ofpbuf *buffer, int in_port, int flood) } static void -output_packet(struct datapath *dp, struct ofpbuf *buffer, uint16_t out_port) +output_packet(struct datapath *dp, struct ofpbuf *buffer, uint16_t out_port, + uint32_t queue_id) { - struct sw_port *p = lookup_port(dp, out_port); + uint16_t class_id; + struct sw_queue * q; + struct sw_port *p; + + q = NULL; + p = dp_lookup_port(dp, out_port); if (p && p->netdev != NULL) { if (!(p->config & OFPPC_PORT_DOWN)) { - if (!netdev_send(p->netdev, buffer)) { + /* avoid the queue lookup for best-effort traffic */ + if (queue_id == 0) { + class_id = 0; + } + else { + /* silently drop the packet if queue doesn't exist */ + q = dp_lookup_queue(p, queue_id); + if (q) { + class_id = q->class_id; + } + else { + goto error; + } + } + + if (!netdev_send(p->netdev, buffer, class_id)) { p->tx_packets++; p->tx_bytes += buffer->size; + if (q) { + q->tx_packets++; + q->tx_bytes += buffer->size; + } } else { p->tx_dropped++; } @@ -528,27 +578,30 @@ output_packet(struct datapath *dp, struct ofpbuf *buffer, uint16_t out_port) return; } + error: ofpbuf_delete(buffer); - VLOG_DBG_RL(&rl, "can't forward to bad port %d\n", out_port); + VLOG_DBG_RL(&rl, "can't forward to bad port:queue(%d:%d)\n", out_port, + queue_id); } -/* Takes ownership of 'buffer' and transmits it to 'out_port' on 'dp'. +/** Takes ownership of 'buffer' and transmits it to 'out_port' on 'dp'. */ void dp_output_port(struct datapath *dp, struct ofpbuf *buffer, - int in_port, int out_port, bool ignore_no_fwd UNUSED) + int in_port, int out_port, uint32_t queue_id, + bool ignore_no_fwd UNUSED) { assert(buffer); switch (out_port) { case OFPP_IN_PORT: - output_packet(dp, buffer, in_port); + output_packet(dp, buffer, in_port, queue_id); break; case OFPP_TABLE: { - struct sw_port *p = lookup_port(dp, in_port); - if (run_flow_through_tables(dp, buffer, p)) { - ofpbuf_delete(buffer); + struct sw_port *p = dp_lookup_port(dp, in_port); + if (run_flow_through_tables(dp, buffer, p)) { + ofpbuf_delete(buffer); } break; } @@ -571,7 +624,7 @@ dp_output_port(struct datapath *dp, struct ofpbuf *buffer, VLOG_DBG_RL(&rl, "can't directly forward to input port"); return; } - output_packet(dp, buffer, out_port); + output_packet(dp, buffer, out_port, queue_id); break; } } @@ -654,6 +707,27 @@ dp_output_control(struct datapath *dp, struct ofpbuf *buffer, int in_port, send_openflow_buffer(dp, buffer, NULL); } +static void +fill_queue_desc(struct ofpbuf *buffer, struct sw_queue *q, + struct ofp_packet_queue *desc) +{ + struct ofp_queue_prop_min_rate *mr; + int len; + + len = sizeof(struct ofp_packet_queue) + + sizeof(struct ofp_queue_prop_min_rate); + desc->queue_id = htonl(q->queue_id); + desc->len = htons(len); + + /* Property list */ + mr = ofpbuf_put_zeros(buffer, sizeof *mr); + mr->prop_header.property = htons(OFPQT_MIN_RATE); + len = sizeof(struct ofp_queue_prop_min_rate); + mr->prop_header.len = htons(len); + mr->rate = htons(q->min_rate); +} + + static void fill_port_desc(struct sw_port *p, struct ofp_phy_port *desc) { @@ -665,9 +739,9 @@ fill_port_desc(struct sw_port *p, struct ofp_phy_port *desc) desc->config = htonl(p->config); desc->state = htonl(p->state); desc->curr = htonl(netdev_get_features(p->netdev, NETDEV_FEAT_CURRENT)); - desc->supported = htonl(netdev_get_features(p->netdev, + desc->supported = htonl(netdev_get_features(p->netdev, NETDEV_FEAT_SUPPORTED)); - desc->advertised = htonl(netdev_get_features(p->netdev, + desc->advertised = htonl(netdev_get_features(p->netdev, NETDEV_FEAT_ADVERTISED)); desc->peer = htonl(netdev_get_features(p->netdev, NETDEV_FEAT_PEER)); } @@ -681,7 +755,7 @@ dp_send_features_reply(struct datapath *dp, const struct sender *sender) ofr = make_openflow_reply(sizeof *ofr, OFPT_FEATURES_REPLY, sender, &buffer); - ofr->datapath_id = htonll(dp->id); + ofr->datapath_id = htonll(dp->id); ofr->n_tables = dp->chain->n_tables; ofr->n_buffers = htonl(N_PKT_BUFFERS); ofr->capabilities = htonl(OFP_SUPPORTED_CAPABILITIES); @@ -697,7 +771,7 @@ dp_send_features_reply(struct datapath *dp, const struct sender *sender) void update_port_flags(struct datapath *dp, const struct ofp_port_mod *opm) { - struct sw_port *p = lookup_port(dp, ntohs(opm->port_no)); + struct sw_port *p = dp_lookup_port(dp, ntohs(opm->port_no)); /* Make sure the port id hasn't changed since this was sent */ if (!p || memcmp(opm->hw_addr, netdev_get_etheraddr(p->netdev), @@ -714,7 +788,7 @@ update_port_flags(struct datapath *dp, const struct ofp_port_mod *opm) } static void -send_port_status(struct sw_port *p, uint8_t status) +send_port_status(struct sw_port *p, uint8_t status) { struct ofpbuf *buffer; struct ofp_port_status *ops; @@ -834,7 +908,7 @@ int run_flow_through_tables(struct datapath *dp, struct ofpbuf *buffer, flow = chain_lookup(dp->chain, &key, 0); if (flow != NULL) { flow_used(flow, buffer); - execute_actions(dp, buffer, &key, flow->sf_acts->actions, + execute_actions(dp, buffer, &key, flow->sf_acts->actions, flow->sf_acts->actions_len, false); return 0; } else { @@ -936,7 +1010,7 @@ recv_packet_out(struct datapath *dp, const struct sender *sender, } else { buffer = retrieve_buffer(ntohl(opo->buffer_id)); if (!buffer) { - return -ESRCH; + return -ESRCH; } } @@ -975,7 +1049,7 @@ add_flow(struct datapath *dp, const struct sender *sender, { int error = -ENOMEM; uint16_t v_code; - struct sw_flow *flow; + struct sw_flow *flow; size_t actions_len = ntohs(ofm->header.length) - sizeof *ofm; int overlap; @@ -1027,7 +1101,7 @@ add_flow(struct datapath *dp, const struct sender *sender, error = chain_insert(dp->chain, flow, (ntohs(ofm->flags) & OFPFF_EMERG) ? 1 : 0); if (error == -ENOBUFS) { - dp_send_error_msg(dp, sender, OFPET_FLOW_MOD_FAILED, + dp_send_error_msg(dp, sender, OFPET_FLOW_MOD_FAILED, OFPFMFC_ALL_TABLES_FULL, ofm, ntohs(ofm->header.length)); goto error_free_flow; } else if (error) { @@ -1042,10 +1116,10 @@ add_flow(struct datapath *dp, const struct sender *sender, uint16_t in_port = ntohs(ofm->match.in_port); flow_extract(buffer, in_port, &key.flow); flow_used(flow, buffer); - execute_actions(dp, buffer, &key, + execute_actions(dp, buffer, &key, ofm->actions, actions_len, false); } else { - error = -ESRCH; + error = -ESRCH; } } return error; @@ -1227,7 +1301,7 @@ static int flow_stats_dump(struct datapath *dp, void *state, { struct sw_table *table = dp->chain->tables[s->table_idx]; - if (table->iterate(table, &match_key, s->rq.out_port, + if (table->iterate(table, &match_key, s->rq.out_port, &s->position, flow_stats_dump_callback, s)) break; @@ -1297,7 +1371,7 @@ static int aggregate_stats_dump(struct datapath *dp, void *state, { struct sw_table *table = dp->chain->tables[table_idx]; - error = table->iterate(table, &match_key, rq->out_port, &position, + error = table->iterate(table, &match_key, rq->out_port, &position, aggregate_stats_dump_callback, rpy); if (error) return error; @@ -1313,7 +1387,7 @@ static int aggregate_stats_dump(struct datapath *dp, void *state, return 0; } -static void aggregate_stats_done(void *state) +static void aggregate_stats_done(void *state) { free(state); } @@ -1343,6 +1417,11 @@ struct port_stats_state { int port; }; +struct queue_stats_state { + uint16_t port; + uint32_t queue_id; +}; + static int port_stats_init(const void *body UNUSED, int body_len UNUSED, void **state) { @@ -1398,6 +1477,79 @@ static void port_stats_done(void *state) free(state); } +static int +queue_stats_init(const void *body, int body_len UNUSED, void **state) +{ + const struct ofp_queue_stats_request *qsr = body; + struct queue_stats_state *s = xmalloc(sizeof *s); + s->port = ntohs(qsr->port_no); + s->queue_id = ntohl(qsr->queue_id); + *state = s; + return 0; +} + +static void +dump_queue_stats(struct sw_queue *q, struct ofpbuf *buffer) +{ + struct ofp_queue_stats *oqs = ofpbuf_put_uninit(buffer, sizeof *oqs); + oqs->port_no = htons(q->port->port_no); + oqs->queue_id = htonl(q->queue_id); + oqs->tx_bytes = htonll(q->tx_bytes); + oqs->tx_packets = htonll(q->tx_packets); + oqs->tx_errors = htonll(q->tx_errors); +} + +static int +queue_stats_dump(struct datapath *dp, void *state, + struct ofpbuf *buffer) +{ + struct queue_stats_state *s = state; + struct sw_queue *q; + struct sw_port *p; + + + if (s->port == OFPP_ALL) { + LIST_FOR_EACH(p, struct sw_port, node, &dp->port_list) { + if (p->port_no < OFPP_MAX) { + if (s->queue_id == OFPQ_ALL) { + LIST_FOR_EACH(q, struct sw_queue, node, &p->queue_list) { + dump_queue_stats(q,buffer); + } + } + else { + q = dp_lookup_queue(p, s->queue_id); + if (q) { + dump_queue_stats(q, buffer); + } + } + } + } + } + else { + p = dp_lookup_port(dp, s->port); + if (p) { + if (s->queue_id == OFPQ_ALL) { + LIST_FOR_EACH(q, struct sw_queue, node, &p->queue_list) { + dump_queue_stats(q,buffer); + } + } + else { + q = dp_lookup_queue(p, s->queue_id); + if (q) { + dump_queue_stats(q, buffer); + } + } + } + } + return 0; +} + +static void +queue_stats_done(void *state) +{ + free(state); +} + /* * We don't define any vendor_stats_state, we let the actual * vendor implementation do that. @@ -1524,6 +1676,14 @@ static const struct stats_type stats[] = { port_stats_dump, port_stats_done }, + { + OFPST_QUEUE, + sizeof(struct ofp_queue_stats_request), + sizeof(struct ofp_queue_stats_request), + queue_stats_init, + queue_stats_dump, + queue_stats_done + }, { OFPST_VENDOR, 8, /* vendor + subtype */ @@ -1621,7 +1781,7 @@ recv_stats_request(struct datapath *dp UNUSED, const struct sender *sender, cb->sender = *sender; cb->s = st; cb->state = NULL; - + body_len = rq_len - offsetof(struct ofp_stats_request, body); if (body_len < cb->s->min_body || body_len > cb->s->max_body) { VLOG_WARN_RL(&rl, "stats request type %d with bad body length %d", @@ -1663,17 +1823,61 @@ recv_echo_reply(struct datapath *dp UNUSED, const struct sender *sender UNUSED, return 0; } +static int +recv_queue_get_config_request(struct datapath *dp, const struct sender *sender, + const void *oh) +{ + struct ofpbuf *buffer; + struct ofp_queue_get_config_reply *ofq_reply; + const struct ofp_queue_get_config_request *ofq_request; + struct sw_port *p; + struct sw_queue *q; + uint16_t port_no; + + ofq_request = (struct ofp_queue_get_config_request *)oh; + port_no = ntohs(ofq_request->port); + + if (port_no < OFPP_MAX) { + /* Find port under query */ + p = dp_lookup_port(dp,port_no); + + /* if the port under query doesn't exist, send an error */ + if (!p || (p->port_no != port_no)) { + dp_send_error_msg(dp, sender, OFPET_QUEUE_OP_FAILED, OFPQOFC_BAD_PORT, + oh, ntohs(ofq_request->header.length)); + goto error; + } + ofq_reply = make_openflow_reply(sizeof *ofq_reply, OFPT_QUEUE_GET_CONFIG_REPLY, + sender, &buffer); + ofq_reply->port = htons(port_no); + LIST_FOR_EACH(q, struct sw_queue, node, &p->queue_list) { + struct ofp_packet_queue * opq = ofpbuf_put_zeros(buffer, sizeof *opq); + fill_queue_desc(buffer, q, opq); + } + send_openflow_buffer(dp, buffer, sender); + } + else { + dp_send_error_msg(dp, sender, OFPET_QUEUE_OP_FAILED, OFPQOFC_BAD_PORT, + oh, ntohs(ofq_request->header.length)); + } + error: + return 0; +} + static int recv_vendor(struct datapath *dp, const struct sender *sender, const void *oh) { const struct ofp_vendor_header *ovh = oh; - switch (ntohl(ovh->vendor)) + switch (ntohl(ovh->vendor)) { case PRIVATE_VENDOR_ID: return private_recv_msg(dp, sender, oh); + case OPENFLOW_VENDOR_ID: + return of_ext_recv_msg(dp, sender, oh); + default: VLOG_WARN_RL(&rl, "unknown vendor: 0x%x\n", ntohl(ovh->vendor)); dp_send_error_msg(dp, sender, OFPET_BAD_REQUEST, @@ -1741,6 +1945,10 @@ fwd_control_input(struct datapath *dp, const struct sender *sender, min_size = sizeof(struct ofp_header); handler = recv_echo_reply; break; + case OFPT_QUEUE_GET_CONFIG_REQUEST: + min_size = sizeof(struct ofp_header); + handler = recv_queue_get_config_request; + break; case OFPT_VENDOR: min_size = sizeof(struct ofp_vendor_header); handler = recv_vendor; @@ -1783,7 +1991,7 @@ uint32_t save_buffer(struct ofpbuf *buffer) if (time_now() < p->timeout) { /* FIXME */ return (uint32_t)-1; } else { - ofpbuf_delete(p->buffer); + ofpbuf_delete(p->buffer); } } /* Don't use maximum cookie value since the all-bits-1 id is diff --git a/udatapath/datapath.h b/udatapath/datapath.h index 136bfe4..113604f 100644 --- a/udatapath/datapath.h +++ b/udatapath/datapath.h @@ -42,12 +42,26 @@ #include "ofpbuf.h" #include "timeval.h" #include "list.h" +#include "netdev.h" struct rconn; struct pvconn; struct sw_flow; struct sender; +struct sw_queue { + struct list node; /* element in port.queues */ + unsigned long long int tx_packets; + unsigned long long int tx_bytes; + unsigned long long int tx_errors; + uint32_t queue_id; + uint16_t class_id; /* internal mapping from OF queue_id to tc class_id */ + struct sw_port *port; /* reference to the parent port */ + /* keep it simple for now, only one property (assuming min_rate) */ + uint16_t property; /* one from OFPQT_ */ + uint16_t min_rate; +}; + struct sw_port { uint32_t config; /* Some subset of OFPPC_* flags. */ uint32_t state; /* Some subset of OFPPS_* flags. */ @@ -58,6 +72,9 @@ struct sw_port { unsigned long long int rx_bytes, tx_bytes; unsigned long long int tx_dropped; uint16_t port_no; + /* port queues */ + struct sw_queue queues[NETDEV_MAX_QUEUES]; + struct list queue_list; /* list of all queues for this port */ }; #define DP_MAX_PORTS 255 @@ -99,8 +116,10 @@ void dp_send_error_msg(struct datapath *, const struct sender *, void dp_send_flow_end(struct datapath *, struct sw_flow *, enum ofp_flow_removed_reason); void dp_output_port(struct datapath *, struct ofpbuf *, int in_port, - int out_port, bool ignore_no_fwd); + int out_port, uint32_t queue_id, bool ignore_no_fwd); void dp_output_control(struct datapath *, struct ofpbuf *, int in_port, size_t max_len, int reason); +struct sw_port * dp_lookup_port(struct datapath *, uint16_t); +struct sw_queue * dp_lookup_queue(struct sw_port *, uint32_t); #endif /* datapath.h */ diff --git a/udatapath/dp_act.c b/udatapath/dp_act.c index 45bf690..28e9590 100644 --- a/udatapath/dp_act.c +++ b/udatapath/dp_act.c @@ -39,7 +39,6 @@ #include "dp_act.h" #include "openflow/nicira-ext.h" - static uint16_t validate_output(struct datapath *dp UNUSED, const struct sw_flow_key *key, const struct ofp_action_header *ah) @@ -57,12 +56,26 @@ validate_output(struct datapath *dp UNUSED, const struct sw_flow_key *key, return ACT_VALIDATION_OK; } +static uint16_t +validate_queue(struct datapath *dp UNUSED, const struct sw_flow_key *key, + const struct ofp_action_header *ah) +{ + struct ofp_action_enqueue *ea = (struct ofp_action_enqueue *)ah; + + /* Only physical ports may have queues. */ + if (ntohs(ea->port) > OFPP_MAX && ntohs(ea->port) != OFPP_IN_PORT) { + return OFPBAC_BAD_OUT_PORT; + } + return ACT_VALIDATION_OK; +} + static void do_output(struct datapath *dp, struct ofpbuf *buffer, int in_port, - size_t max_len, int out_port, bool ignore_no_fwd) + size_t max_len, int out_port, uint32_t queue_id, + bool ignore_no_fwd) { if (out_port != OFPP_CONTROLLER) { - dp_output_port(dp, buffer, in_port, out_port, ignore_no_fwd); + dp_output_port(dp, buffer, in_port, out_port, queue_id, ignore_no_fwd); } else { dp_output_control(dp, buffer, in_port, max_len, OFPR_ACTION); } @@ -267,6 +280,12 @@ static const struct openflow_action of_actions[] = { validate_output, NULL /* This is optimized into execute_actions */ }, + [OFPAT_ENQUEUE] = { + sizeof(struct ofp_action_enqueue), + sizeof(struct ofp_action_enqueue), + validate_queue, + NULL /* This is optimized into execute_actions */ + }, [OFPAT_SET_VLAN_VID] = { sizeof(struct ofp_action_vlan_vid), sizeof(struct ofp_action_vlan_vid), @@ -461,11 +480,13 @@ void execute_actions(struct datapath *dp, struct ofpbuf *buffer, * freeing the original buffer is wasteful. So the following code is * slightly obscure just to avoid that. */ int prev_port; + uint32_t prev_queue; size_t max_len = UINT16_MAX; uint16_t in_port = ntohs(key->flow.in_port); uint8_t *p = (uint8_t *)actions; prev_port = -1; + prev_queue = 0; /* The action list was already validated, so we can be a bit looser * in our sanity-checking. */ @@ -475,14 +496,20 @@ void execute_actions(struct datapath *dp, struct ofpbuf *buffer, if (prev_port != -1) { do_output(dp, ofpbuf_clone(buffer), in_port, max_len, - prev_port, ignore_no_fwd); + prev_port, prev_queue, ignore_no_fwd); prev_port = -1; } if (ah->type == htons(OFPAT_OUTPUT)) { struct ofp_action_output *oa = (struct ofp_action_output *)p; prev_port = ntohs(oa->port); + prev_queue = 0; /* using the default best-effort queue */ max_len = ntohs(oa->max_len); + } else if (ah->type == htons(OFPAT_ENQUEUE)) { + struct ofp_action_enqueue *ea = (struct ofp_action_enqueue *)p; + prev_port = ntohs(ea->port); + prev_queue = ntohl(ea->queue_id); + max_len = 0; /* we will not send to the controller anyways - useless */ } else { uint16_t type = ntohs(ah->type); @@ -497,7 +524,7 @@ void execute_actions(struct datapath *dp, struct ofpbuf *buffer, actions_len -= len; } if (prev_port != -1) { - do_output(dp, buffer, in_port, max_len, prev_port, ignore_no_fwd); + do_output(dp, buffer, in_port, max_len, prev_port, prev_queue, ignore_no_fwd); } else { ofpbuf_delete(buffer); }