X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Frconn.c;h=46fd51df35217337f164a2ea37dd91b99047f3ef;hb=d6ef9d8adea5fdf5a8be3ad5c718cea234f0b998;hp=73d42bb91cc4bee1220c902c9e407c6fea89fc08;hpb=1daf19850e40d23d7d87d8a7b6417d9f3faab386;p=sliver-openvswitch.git

diff --git a/lib/rconn.c b/lib/rconn.c
index 73d42bb91..46fd51df3 100644
--- a/lib/rconn.c
+++ b/lib/rconn.c
@@ -57,21 +57,38 @@ struct rconn {
     int txq_limit;
     time_t backoff_deadline;
     int backoff;
+    time_t last_connected;
+    unsigned int packets_sent;
+
+    /* Throughout this file, "probe" is shorthand for "inactivity probe".
+     * When nothing has been received from the peer for a while, we send out
+     * an echo request as an inactivity probe packet.  We should receive back
+     * a response. */
+    int probe_interval;         /* Secs of inactivity before sending probe. */
+    time_t probe_sent;          /* Time at which last probe sent, or 0 if none
+                                 * has been sent since 'last_connected'. */
 };
 
 static struct rconn *create_rconn(const char *name, int txq_limit,
-                                  struct vconn *);
+                                  int probe_interval, struct vconn *);
 static int try_send(struct rconn *);
 static void disconnect(struct rconn *, int error);
+static time_t probe_deadline(const struct rconn *);
 
 /* Creates and returns a new rconn that connects (and re-connects as necessary)
  * to the vconn named 'name'.
  *
- * 'txq_limit' is the maximum length of the send queue, in packets. */
+ * 'txq_limit' is the maximum length of the send queue, in packets.
+ *
+ * 'probe_interval' is a number of seconds.  If the interval passes once
+ * without an OpenFlow message being received from the peer, the rconn sends
+ * out an "echo request" message.  If the interval passes again without a
+ * message being received, the rconn disconnects and re-connects to the peer.
+ * Setting 'probe_interval' to 0 disables this behavior.  */
 struct rconn *
-rconn_new(const char *name, int txq_limit) 
+rconn_new(const char *name, int txq_limit, int probe_interval)
 {
-    return create_rconn(name, txq_limit, NULL);
+    return create_rconn(name, txq_limit, probe_interval, NULL);
 }
 
 /* Creates and returns a new rconn that is initially connected to 'vconn' and
@@ -83,7 +100,7 @@ struct rconn *
 rconn_new_from_vconn(const char *name, int txq_limit, struct vconn *vconn)
 {
     assert(vconn != NULL);
-    return create_rconn(name, txq_limit, vconn);
+    return create_rconn(name, txq_limit, 0, vconn);
 }
 
 /* Disconnects 'rc' and frees the underlying storage. */
@@ -132,6 +149,22 @@ rconn_run(struct rconn *rc)
             disconnect(rc, 0);
         }
     } else {
+        if (rc->probe_interval) {
+            time_t now = time(0);
+            if (now >= probe_deadline(rc)) {
+                if (!rc->probe_sent) {
+                    queue_push_tail(&rc->txq, make_echo_request());
+                    rc->probe_sent = now;
+                    VLOG_DBG("%s: idle %d seconds, sending inactivity probe",
+                             rc->name, (int) (now - rc->last_connected)); 
+                } else {
+                    VLOG_ERR("%s: no response to inactivity probe after %d "
+                             "seconds, disconnecting",
+                             rc->name, (int) (now - rc->probe_sent));
+                    disconnect(rc, 0);
+                }
+            }
+        }
         while (rc->txq.n > 0) {
             int error = try_send(rc);
             if (error == EAGAIN) {
@@ -147,17 +180,31 @@ rconn_run(struct rconn *rc)
 /* Causes the next call to poll_block() to wake up when rconn_run() should be
  * called on 'rc'. */
 void
-rconn_run_wait(struct rconn *rc) 
+rconn_run_wait(struct rconn *rc)
 {
     if (rc->vconn) {
         if (rc->txq.n) {
             vconn_wait(rc->vconn, WAIT_SEND);
         }
+        if (rc->probe_interval) {
+            poll_timer_wait((probe_deadline(rc) - time(0)) * 1000);
+        }
     } else {
         poll_timer_wait((rc->backoff_deadline - time(0)) * 1000);
     }
 }
 
+/* Returns the time at which, should nothing be received, we should send out an
+ * inactivity probe (if none has yet been sent) or conclude that the connection
+ * is dead (if a probe has already been sent). */
+static time_t
+probe_deadline(const struct rconn *rc) 
+{
+    assert(rc->probe_interval);
+    return (rc->probe_interval
+            + (rc->probe_sent ? rc->probe_sent : rc->last_connected));
+}
+
 /* Attempts to receive a packet from 'rc'.  If successful, returns the packet;
  * otherwise, returns a null pointer.  The caller is responsible for freeing
  * the packet (with buffer_delete()). */
@@ -168,6 +215,8 @@ rconn_recv(struct rconn *rc)
         struct buffer *buffer;
         int error = vconn_recv(rc->vconn, &buffer);
         if (!error) {
+            rc->last_connected = time(0);
+            rc->probe_sent = 0;
             return buffer;
         } else if (error != EAGAIN) {
             disconnect(rc, error); 
@@ -186,14 +235,13 @@ rconn_recv_wait(struct rconn *rc)
     }
 }
 
-/* There is no rconn_send_wait() function: an rconn has a send queue that it
- * takes care of sending if you call rconn_wait(), which will have the side
- * effect of waking up poll_block(). */
+/* Sends 'b' on 'rc'.  Returns 0 if successful, EAGAIN if at least 'txq_limit'
+ * packets are already queued, otherwise a positive errno value. */
 int
-rconn_send(struct rconn *rc, struct buffer *b) 
+do_send(struct rconn *rc, struct buffer *b, int txq_limit)
 {
     if (rc->vconn) {
-        if (rc->txq.n < rc->txq_limit) {
+        if (rc->txq.n < txq_limit) {
             queue_push_tail(&rc->txq, b);
             if (rc->txq.n == 1) {
                 try_send(rc);
@@ -207,6 +255,46 @@ rconn_send(struct rconn *rc, struct buffer *b)
     }
 }
 
+/* Sends 'b' on 'rc'.  Returns 0 if successful, EAGAIN if the send queue is
+ * full, or ENOTCONN if 'rc' is not currently connected.
+ *
+ * There is no rconn_send_wait() function: an rconn has a send queue that it
+ * takes care of sending if you call rconn_run(), which will have the side
+ * effect of waking up poll_block(). */
+int
+rconn_send(struct rconn *rc, struct buffer *b)
+{
+    return do_send(rc, b, rc->txq_limit);
+}
+
+/* Sends 'b' on 'rc'.  Returns 0 if successful, EAGAIN if the send queue is
+ * full, otherwise a positive errno value.
+ *
+ * Compared to rconn_send(), this function relaxes the queue limit, allowing
+ * more packets than usual to be queued. */
+int
+rconn_force_send(struct rconn *rc, struct buffer *b)
+{
+    return do_send(rc, b, 2 * rc->txq_limit);
+}
+
+/* Returns true if 'rc''s send buffer is full,
+ * false if it has room for at least one more packet. */
+bool
+rconn_is_full(const struct rconn *rc) 
+{
+    return rc->txq.n >= rc->txq_limit;
+}
+
+/* Returns the total number of packets successfully sent on the underlying
+ * vconn.  A packet is not counted as sent while it is still queued in the
+ * rconn, only when it has been successfuly passed to the vconn.  */
+unsigned int
+rconn_packets_sent(const struct rconn *rc) 
+{
+    return rc->packets_sent;
+}
+
 /* Returns 'rc''s name (the 'name' argument passed to rconn_new()). */
 const char *
 rconn_get_name(const struct rconn *rc) 
@@ -221,9 +309,25 @@ rconn_is_alive(const struct rconn *rconn)
 {
     return rconn->reliable || rconn->vconn;
 }
+
+/* Returns true if 'rconn' is connected, false otherwise. */
+bool
+rconn_is_connected(const struct rconn *rconn)
+{
+    return rconn->vconn && !vconn_connect(rconn->vconn);
+}
+
+/* Returns 0 if 'rconn' is connected, otherwise the number of seconds that it
+ * has been disconnected. */
+int
+rconn_disconnected_duration(const struct rconn *rconn) 
+{
+    return rconn_is_connected(rconn) ? 0 : time(0) - rconn->last_connected;
+}
 
 static struct rconn *
-create_rconn(const char *name, int txq_limit, struct vconn *vconn)
+create_rconn(const char *name, int txq_limit, int probe_interval,
+             struct vconn *vconn)
 {
     struct rconn *rc = xmalloc(sizeof *rc);
     assert(txq_limit > 0);
@@ -235,6 +339,11 @@ create_rconn(const char *name, int txq_limit, struct vconn *vconn)
     rc->txq_limit = txq_limit;
     rc->backoff_deadline = 0;
     rc->backoff = 0;
+    rc->last_connected = time(0);
+    rc->probe_interval = (probe_interval
+                              ? MAX(5, probe_interval) : 0);
+    rc->probe_sent = 0;
+    rc->packets_sent = 0;
     return rc;
 }
 
@@ -249,6 +358,7 @@ try_send(struct rconn *rc)
     if (retval) {
         return retval;
     }
+    rc->packets_sent++;
     queue_advance_head(&rc->txq, next);
     return 0;
 }
@@ -285,4 +395,5 @@ disconnect(struct rconn *rc, int error)
                   rc->name, rc->backoff);
     }
     rc->backoff_deadline = now + rc->backoff;
+    rc->probe_sent = 0;
 }