Drop rconn's responsibility for limiting the tx queue.
[sliver-openvswitch.git] / switch / datapath.c
index 3d639ff..1d10ed4 100644 (file)
@@ -40,6 +40,7 @@
 #include <string.h>
 #include "buffer.h"
 #include "chain.h"
+#include "csum.h"
 #include "flow.h"
 #include "netdev.h"
 #include "packets.h"
@@ -47,6 +48,7 @@
 #include "rconn.h"
 #include "vconn.h"
 #include "table.h"
+#include "timeval.h"
 #include "xtoxll.h"
 
 #define THIS_MODULE VLM_datapath
@@ -88,6 +90,8 @@ struct sender {
 struct remote {
     struct list node;
     struct rconn *rconn;
+#define TXQ_LIMIT 128           /* Max number of packets to queue for tx. */
+    int n_txq;                  /* Number of packets queued for tx on rconn. */
 
     /* Support for reliable, multi-message replies to requests.
      *
@@ -112,7 +116,9 @@ struct datapath {
 
     struct sw_chain *chain;  /* Forwarding rules. */
 
-    struct ofp_switch_config config;
+    /* Configuration set from controller. */
+    uint16_t flags;
+    uint16_t miss_send_len;
 
     /* Switch ports. */
     struct sw_port ports[OFPP_MAX];
@@ -129,7 +135,8 @@ void dp_output_port(struct datapath *, struct buffer *,
 void dp_update_port_flags(struct datapath *dp, const struct ofp_phy_port *opp);
 void dp_output_control(struct datapath *, struct buffer *, int in_port,
                        size_t max_len, int reason);
-static void send_flow_expired(struct datapath *, struct sw_flow *);
+static void send_flow_expired(struct datapath *, struct sw_flow *,
+                              enum ofp_flow_expired_reason);
 static void send_port_status(struct sw_port *p, uint8_t status);
 static void del_switch_port(struct sw_port *p);
 static void execute_actions(struct datapath *, struct buffer *,
@@ -153,6 +160,7 @@ static void modify_th(struct buffer *buffer, uint16_t eth_proto,
 
 #define PKT_COOKIE_BITS (32 - PKT_BUFFER_BITS)
 
+int run_flow_through_tables(struct datapath *, struct buffer *, int in_port);
 void fwd_port_input(struct datapath *, struct buffer *, int in_port);
 int fwd_control_input(struct datapath *, const struct sender *,
                       const void *, size_t);
@@ -186,7 +194,7 @@ dp_new(struct datapath **dp_, uint64_t dpid, struct rconn *rconn)
         return ENOMEM;
     }
 
-    dp->last_timeout = time(0);
+    dp->last_timeout = time_now();
     list_init(&dp->remotes);
     dp->controller = remote_create(dp, rconn);
     dp->listen_vconn = NULL;
@@ -199,8 +207,8 @@ dp_new(struct datapath **dp_, uint64_t dpid, struct rconn *rconn)
     }
 
     list_init(&dp->port_list);
-    dp->config.flags = 0;
-    dp->config.miss_send_len = htons(OFP_DEFAULT_MISS_SEND_LEN);
+    dp->flags = 0;
+    dp->miss_send_len = OFP_DEFAULT_MISS_SEND_LEN;
     *dp_ = dp;
     return 0;
 }
@@ -214,11 +222,11 @@ dp_add_port(struct datapath *dp, const char *name)
     struct sw_port *p;
     int error;
 
-    error = netdev_open(name, &netdev);
+    error = netdev_open(name, NETDEV_ETH_TYPE_ANY, &netdev);
     if (error) {
         return error;
     }
-    error = netdev_set_flags(netdev, NETDEV_UP | NETDEV_PROMISC);
+    error = netdev_set_flags(netdev, NETDEV_UP | NETDEV_PROMISC, false);
     if (error) {
         VLOG_ERR("Couldn't set promiscuous mode on %s device", name);
         netdev_close(netdev);
@@ -264,7 +272,7 @@ dp_add_listen_vconn(struct datapath *dp, struct vconn *listen_vconn)
 void
 dp_run(struct datapath *dp)
 {
-    time_t now = time(0);
+    time_t now = time_now();
     struct sw_port *p, *pn;
     struct remote *r, *rn;
     struct buffer *buffer = NULL;
@@ -275,7 +283,7 @@ dp_run(struct datapath *dp)
 
         chain_timeout(dp->chain, &deleted);
         LIST_FOR_EACH_SAFE (f, n, struct sw_flow, node, &deleted) {
-            send_flow_expired(dp, f);
+            send_flow_expired(dp, f, f->reason);
             list_remove(&f->node);
             flow_free(f);
         }
@@ -325,7 +333,7 @@ dp_run(struct datapath *dp)
                 }
                 break;
             }
-            remote_create(dp, rconn_new_from_vconn("passive", 128, new_vconn));
+            remote_create(dp, rconn_new_from_vconn("passive", new_vconn));
         }
     }
 }
@@ -361,7 +369,7 @@ remote_run(struct datapath *dp, struct remote *r)
             }
             buffer_delete(buffer); 
         } else {
-            if (!rconn_is_full(r->rconn)) {
+            if (r->n_txq < TXQ_LIMIT) {
                 int error = r->cb_dump(dp, r->cb_aux);
                 if (error <= 0) {
                     if (error) {
@@ -545,16 +553,8 @@ dp_output_port(struct datapath *dp, struct buffer *buffer,
     } else if (out_port == OFPP_CONTROLLER) {
         dp_output_control(dp, buffer, in_port, 0, OFPR_ACTION); 
     } else if (out_port == OFPP_TABLE) {
-        struct sw_flow_key key;
-        struct sw_flow *flow;
-
-        key.wildcards = 0;
-        flow_extract(buffer, in_port, &key.flow);
-        flow = chain_lookup(dp->chain, &key);
-        if (flow != NULL) {
-            flow_used(flow, buffer);
-            execute_actions(dp, buffer, in_port, &key, 
-                            flow->actions, flow->n_actions);
+               if (run_flow_through_tables(dp, buffer, in_port)) {
+                       buffer_delete(buffer);
         }
     } else {
         output_packet(dp, buffer, out_port);
@@ -562,19 +562,11 @@ dp_output_port(struct datapath *dp, struct buffer *buffer,
 }
 
 static void *
-alloc_openflow_buffer(struct datapath *dp, size_t openflow_len, uint8_t type,
-                      const struct sender *sender, struct buffer **bufferp)
+make_openflow_reply(size_t openflow_len, uint8_t type,
+                    const struct sender *sender, struct buffer **bufferp)
 {
-    struct buffer *buffer;
-    struct ofp_header *oh;
-
-    buffer = *bufferp = buffer_new(openflow_len);
-    oh = buffer_put_uninit(buffer, openflow_len);
-    oh->version = OFP_VERSION;
-    oh->type = type;
-    oh->length = 0;             /* Filled in by send_openflow_buffer(). */
-    oh->xid = sender ? sender->xid : 0;
-    return oh;
+    return make_openflow_xid(openflow_len, type, sender ? sender->xid : 0,
+                             bufferp);
 }
 
 static int
@@ -583,13 +575,12 @@ send_openflow_buffer(struct datapath *dp, struct buffer *buffer,
 {
     struct remote *remote = sender ? sender->remote : dp->controller;
     struct rconn *rconn = remote->rconn;
-    struct ofp_header *oh;
     int retval;
 
-    oh = buffer_at_assert(buffer, 0, sizeof *oh);
-    oh->length = htons(buffer->size);
-
-    retval = rconn_send(rconn, buffer);
+    update_openflow_length(buffer);
+    retval = (remote->n_txq < TXQ_LIMIT
+              ? rconn_send(rconn, buffer, &remote->n_txq)
+              : EAGAIN);
     if (retval) {
         VLOG_WARN("send to %s failed: %s",
                   rconn_get_name(rconn), strerror(retval));
@@ -651,8 +642,8 @@ dp_send_features_reply(struct datapath *dp, const struct sender *sender)
     struct ofp_switch_features *ofr;
     struct sw_port *p;
 
-    ofr = alloc_openflow_buffer(dp, sizeof *ofr, OFPT_FEATURES_REPLY,
-                                sender, &buffer);
+    ofr = make_openflow_reply(sizeof *ofr, OFPT_FEATURES_REPLY,
+                               sender, &buffer);
     ofr->datapath_id    = htonll(dp->id); 
     ofr->n_exact        = htonl(2 * TABLE_HASH_MAX_FLOWS);
     ofr->n_compression  = 0;         /* Not supported */
@@ -690,8 +681,7 @@ send_port_status(struct sw_port *p, uint8_t status)
 {
     struct buffer *buffer;
     struct ofp_port_status *ops;
-    ops = alloc_openflow_buffer(p->dp, sizeof *ops, OFPT_PORT_STATUS, NULL,
-                                &buffer);
+    ops = make_openflow_xid(sizeof *ops, OFPT_PORT_STATUS, 0, &buffer);
     ops->reason = status;
     memset(ops->pad, 0, sizeof ops->pad);
     fill_port_desc(p->dp, p, &ops->desc);
@@ -700,18 +690,20 @@ send_port_status(struct sw_port *p, uint8_t status)
 }
 
 void
-send_flow_expired(struct datapath *dp, struct sw_flow *flow)
+send_flow_expired(struct datapath *dp, struct sw_flow *flow,
+                  enum ofp_flow_expired_reason reason)
 {
     struct buffer *buffer;
     struct ofp_flow_expired *ofe;
-    ofe = alloc_openflow_buffer(dp, sizeof *ofe, OFPT_FLOW_EXPIRED, NULL,
-                                &buffer);
+    ofe = make_openflow_xid(sizeof *ofe, OFPT_FLOW_EXPIRED, 0, &buffer);
     flow_fill_match(&ofe->match, &flow->key);
 
-    memset(ofe->pad, 0, sizeof ofe->pad);
     ofe->priority = htons(flow->priority);
+    ofe->reason = reason;
+    memset(ofe->pad, 0, sizeof ofe->pad);
 
-    ofe->duration     = htonl(flow->timeout - flow->max_idle - flow->created);
+    ofe->duration     = htonl(time_now() - flow->created);
+    memset(ofe->pad2, 0, sizeof ofe->pad2);
     ofe->packet_count = htonll(flow->packet_count);
     ofe->byte_count   = htonll(flow->byte_count);
     send_openflow_buffer(dp, buffer, NULL);
@@ -723,8 +715,8 @@ dp_send_error_msg(struct datapath *dp, const struct sender *sender,
 {
     struct buffer *buffer;
     struct ofp_error_msg *oem;
-    oem = alloc_openflow_buffer(dp, sizeof(*oem)+len, OFPT_ERROR_MSG, 
-                                sender, &buffer);
+    oem = make_openflow_reply(sizeof(*oem)+len, OFPT_ERROR_MSG, 
+                              sender, &buffer);
     oem->type = htons(type);
     oem->code = htons(code);
     memcpy(oem->data, data, len);
@@ -754,31 +746,53 @@ fill_flow_stats(struct buffer *buffer, struct sw_flow *flow,
     ofs->match.tp_src    = flow->key.flow.tp_src;
     ofs->match.tp_dst    = flow->key.flow.tp_dst;
     ofs->duration        = htonl(now - flow->created);
+    ofs->priority        = htons(flow->priority);
+    ofs->idle_timeout    = htons(flow->idle_timeout);
+    ofs->hard_timeout    = htons(flow->hard_timeout);
+    memset(ofs->pad2, 0, sizeof ofs->pad2);
     ofs->packet_count    = htonll(flow->packet_count);
     ofs->byte_count      = htonll(flow->byte_count);
-    ofs->priority        = htons(flow->priority);
-    ofs->max_idle        = htons(flow->max_idle);
     memcpy(ofs->actions, flow->actions,
            sizeof *ofs->actions * flow->n_actions);
 }
 
 \f
 /* 'buffer' was received on 'in_port', a physical switch port between 0 and
- * OFPP_MAX.  Process it according to 'chain'. */
-void fwd_port_input(struct datapath *dp, struct buffer *buffer, int in_port)
+ * OFPP_MAX.  Process it according to 'dp''s flow table.  Returns 0 if
+ * successful, in which case 'buffer' is destroyed, or -ESRCH if there is no
+ * matching flow, in which case 'buffer' still belongs to the caller. */
+int run_flow_through_tables(struct datapath *dp, struct buffer *buffer,
+                            int in_port)
 {
     struct sw_flow_key key;
     struct sw_flow *flow;
 
     key.wildcards = 0;
-    flow_extract(buffer, in_port, &key.flow);
+    if (flow_extract(buffer, in_port, &key.flow)
+        && (dp->flags & OFPC_FRAG_MASK) == OFPC_FRAG_DROP) {
+        /* Drop fragment. */
+        buffer_delete(buffer);
+        return 0;
+    }
+
     flow = chain_lookup(dp->chain, &key);
     if (flow != NULL) {
         flow_used(flow, buffer);
         execute_actions(dp, buffer, in_port, &key,
                         flow->actions, flow->n_actions);
+        return 0;
     } else {
-        dp_output_control(dp, buffer, in_port, ntohs(dp->config.miss_send_len),
+        return -ESRCH;
+    }
+}
+
+/* 'buffer' was received on 'in_port', a physical switch port between 0 and
+ * OFPP_MAX.  Process it according to 'dp''s flow table, sending it up to the
+ * controller if no flow matches.  Takes ownership of 'buffer'. */
+void fwd_port_input(struct datapath *dp, struct buffer *buffer, int in_port) 
+{
+    if (run_flow_through_tables(dp, buffer, in_port)) {
+        dp_output_control(dp, buffer, in_port, dp->miss_send_len,
                           OFPR_NO_MATCH);
     }
 }
@@ -858,34 +872,6 @@ execute_actions(struct datapath *dp, struct buffer *buffer,
         buffer_delete(buffer);
 }
 
-/* Returns the new checksum for a packet in which the checksum field previously
- * contained 'old_csum' and in which a field that contained 'old_u16' was
- * changed to contain 'new_u16'. */
-static uint16_t
-recalc_csum16(uint16_t old_csum, uint16_t old_u16, uint16_t new_u16)
-{
-    /* Ones-complement arithmetic is endian-independent, so this code does not
-     * use htons() or ntohs().
-     *
-     * See RFC 1624 for formula and explanation. */
-    uint16_t hc_complement = ~old_csum;
-    uint16_t m_complement = ~old_u16;
-    uint16_t m_prime = new_u16;
-    uint32_t sum = hc_complement + m_complement + m_prime;
-    uint16_t hc_prime_complement = sum + (sum >> 16);
-    return ~hc_prime_complement;
-}
-
-/* Returns the new checksum for a packet in which the checksum field previously
- * contained 'old_csum' and in which a field that contained 'old_u32' was
- * changed to contain 'new_u32'. */
-static uint16_t
-recalc_csum32(uint16_t old_csum, uint32_t old_u32, uint32_t new_u32)
-{
-    return recalc_csum16(recalc_csum16(old_csum, old_u32, new_u32),
-                         old_u32 >> 16, new_u32 >> 16);
-}
-
 static void modify_nh(struct buffer *buffer, uint16_t eth_proto,
                       uint8_t nw_proto, const struct ofp_action *a)
 {
@@ -994,13 +980,11 @@ recv_get_config_request(struct datapath *dp, const struct sender *sender,
     struct buffer *buffer;
     struct ofp_switch_config *osc;
 
-    osc = alloc_openflow_buffer(dp, sizeof *osc, OFPT_GET_CONFIG_REPLY,
-                                sender, &buffer);
+    osc = make_openflow_reply(sizeof *osc, OFPT_GET_CONFIG_REPLY,
+                              sender, &buffer);
 
-    assert(sizeof *osc == sizeof dp->config);
-    memcpy(((char *)osc) + sizeof osc->header,
-           ((char *)&dp->config) + sizeof dp->config.header,
-           sizeof dp->config - sizeof dp->config.header);
+    osc->flags = htons(dp->flags);
+    osc->miss_send_len = htons(dp->miss_send_len);
 
     return send_openflow_buffer(dp, buffer, sender);
 }
@@ -1010,7 +994,15 @@ recv_set_config(struct datapath *dp, const struct sender *sender UNUSED,
                 const void *msg)
 {
     const struct ofp_switch_config *osc = msg;
-    dp->config = *osc;
+    int flags;
+
+    flags = ntohs(osc->flags) & (OFPC_SEND_FLOW_EXP | OFPC_FRAG_MASK);
+    if ((flags & OFPC_FRAG_MASK) != OFPC_FRAG_NORMAL
+        && (flags & OFPC_FRAG_MASK) != OFPC_FRAG_DROP) {
+        flags = (flags & ~OFPC_FRAG_MASK) | OFPC_FRAG_DROP;
+    }
+    dp->flags = flags;
+    dp->miss_send_len = ntohs(osc->miss_send_len);
     return 0;
 }
 
@@ -1088,11 +1080,11 @@ add_flow(struct datapath *dp, const struct ofp_flow_mod *ofm)
 
     /* Fill out flow. */
     flow_extract_match(&flow->key, &ofm->match);
-    flow->max_idle = ntohs(ofm->max_idle);
     flow->priority = flow->key.wildcards ? ntohs(ofm->priority) : -1;
-    flow->timeout = time(0) + flow->max_idle; /* FIXME */
+    flow->idle_timeout = ntohs(ofm->idle_timeout);
+    flow->hard_timeout = ntohs(ofm->hard_timeout);
+    flow->used = flow->created = time_now();
     flow->n_actions = n_acts;
-    flow->created = time(0);    /* FIXME */
     flow->byte_count = 0;
     flow->packet_count = 0;
     memcpy(flow->actions, ofm->actions, n_acts * sizeof *flow->actions);
@@ -1187,7 +1179,7 @@ static int flow_stats_dump(struct datapath *dp, void *state,
 
     flow_extract_match(&match_key, &s->rq.match);
     s->buffer = buffer;
-    s->now = time(0);
+    s->now = time_now();
     while (s->table_idx < dp->chain->n_tables
            && (s->rq.table_id == 0xff || s->rq.table_id == s->table_idx))
     {
@@ -1287,7 +1279,7 @@ static int table_stats_dump(struct datapath *dp, void *state,
         memset(ots->pad, 0, sizeof ots->pad);
         ots->max_entries = htonl(stats.max_flows);
         ots->active_count = htonl(stats.n_flows);
-        ots->matched_count = htonll(0); /* FIXME */
+        ots->matched_count = htonll(stats.n_matched);
     }
     return 0;
 }
@@ -1409,8 +1401,8 @@ stats_dump(struct datapath *dp, void *cb_)
         return 0;
     }
 
-    osr = alloc_openflow_buffer(dp, sizeof *osr, OFPT_STATS_REPLY, &cb->sender,
-                                &buffer);
+    osr = make_openflow_reply(sizeof *osr, OFPT_STATS_REPLY, &cb->sender,
+                              &buffer);
     osr->type = htons(cb->s - stats);
     osr->flags = 0;
 
@@ -1562,8 +1554,8 @@ fwd_control_input(struct datapath *dp, const struct sender *sender,
     struct ofp_header *oh;
 
     oh = (struct ofp_header *) msg;
-    if (oh->version != OFP_VERSION || oh->type >= ARRAY_SIZE(packets)
-        || ntohs(oh->length) > length)
+    assert(oh->version == OFP_VERSION);
+    if (oh->type >= ARRAY_SIZE(packets) || ntohs(oh->length) > length)
         return -EINVAL;
 
     pkt = &packets[oh->type];
@@ -1598,7 +1590,7 @@ uint32_t save_buffer(struct buffer *buffer)
     if (p->buffer) {
         /* Don't buffer packet if existing entry is less than
          * OVERWRITE_SECS old. */
-        if (time(0) < p->timeout) { /* FIXME */
+        if (time_now() < p->timeout) { /* FIXME */
             return -1;
         } else {
             buffer_delete(p->buffer); 
@@ -1609,7 +1601,7 @@ uint32_t save_buffer(struct buffer *buffer)
     if (++p->cookie >= (1u << PKT_COOKIE_BITS) - 1)
         p->cookie = 0;
     p->buffer = buffer_clone(buffer);      /* FIXME */
-    p->timeout = time(0) + OVERWRITE_SECS; /* FIXME */
+    p->timeout = time_now() + OVERWRITE_SECS; /* FIXME */
     id = buffer_idx | (p->cookie << PKT_BUFFER_BITS);
 
     return id;