Drop rconn's responsibility for limiting the tx queue.

[openvswitch] / switch / datapath.c

diff --git a/switch/datapath.c b/switch/datapath.c
index d0013d847943381edb79e281b6e64748be89aee3..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.
      *
@@ -191,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;
@@ -269,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;
@@ -330,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));
         }
     }
 }
@@ -366,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) {
@@ -575,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));
@@ -697,7 +702,7 @@ send_flow_expired(struct datapath *dp, struct sw_flow *flow,
     ofe->reason = reason;
     memset(ofe->pad, 0, sizeof ofe->pad);
 
-    ofe->duration     = htonl(time(0) - 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);
@@ -1078,7 +1083,7 @@ add_flow(struct datapath *dp, const struct ofp_flow_mod *ofm)
     flow->priority = flow->key.wildcards ? ntohs(ofm->priority) : -1;
     flow->idle_timeout = ntohs(ofm->idle_timeout);
     flow->hard_timeout = ntohs(ofm->hard_timeout);
-    flow->used = flow->created = time(0);
+    flow->used = flow->created = time_now();
     flow->n_actions = n_acts;
     flow->byte_count = 0;
     flow->packet_count = 0;
@@ -1174,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))
     {
@@ -1585,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); 
@@ -1596,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;