Drop rconn's responsibility for limiting the tx queue.

[openvswitch] / switch / datapath.c

diff --git a/switch/datapath.c b/switch/datapath.c
index 4d52dda93c538755da8785c9ee928742d23cb0d1..1d10ed421ff4d1d4ca3c7ac21b1e9f103c66db2f 100644 (file)
--- a/switch/datapath.c
+++ b/switch/datapath.c
@@ -48,6 +48,7 @@
 #include "rconn.h"
 #include "vconn.h"
 #include "table.h"
+#include "timeval.h"
 #include "xtoxll.h"
 
 #define THIS_MODULE VLM_datapath
@@ -89,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.
      *
@@ -132,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 *,
@@ -190,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;
@@ -268,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;
@@ -279,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);
         }
@@ -329,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));
         }
     }
 }
@@ -365,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) {
@@ -574,7 +578,9 @@ send_openflow_buffer(struct datapath *dp, struct buffer *buffer,
     int retval;
 
     update_openflow_length(buffer);
-    retval = rconn_send(rconn, 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));
@@ -684,17 +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 = 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);
@@ -737,10 +746,12 @@ 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);
 }
@@ -1069,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);
@@ -1168,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))
     {
@@ -1579,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); 
@@ -1590,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;