vconn: Convert vconn code to modern OVS structure.

The vconn code is a relative fossil as OVS code goes.  It was written
before we had really figured how code should fit together.  Part of that
history is that it used poll_fd_callback() to register callbacks without
the assistance of other code.  That isn't how the rest of OVS works now;
this code is the only remaining user of that function.

To make it more like the rest of the system, this code gets rid of the use
of poll_fd_callback().  It also adds vconn_run() and vconn_run_wait()
functions and calls to them from the places where they are now required.

diff --git a/lib/rconn.c b/lib/rconn.c
index b6e958e..f2d074a 100644 (file)
--- a/lib/rconn.c
+++ b/lib/rconn.c
@@ -459,6 +459,15 @@ void
 rconn_run(struct rconn *rc)
 {
     int old_state;
+    size_t i;
+
+    if (rc->vconn) {
+        vconn_run(rc->vconn);
+    }
+    for (i = 0; i < rc->n_monitors; i++) {
+        vconn_run(rc->monitors[i]);
+    }
+
     do {
         old_state = rc->state;
         switch (rc->state) {
@@ -476,7 +485,17 @@ rconn_run(struct rconn *rc)
 void
 rconn_run_wait(struct rconn *rc)
 {
-    unsigned int timeo = timeout(rc);
+    unsigned int timeo;
+    size_t i;
+
+    if (rc->vconn) {
+        vconn_run_wait(rc->vconn);
+    }
+    for (i = 0; i < rc->n_monitors; i++) {
+        vconn_run_wait(rc->monitors[i]);
+    }
+
+    timeo = timeout(rc);
     if (timeo != UINT_MAX) {
         unsigned int expires = sat_add(rc->state_entered, timeo);
         unsigned int remaining = sat_sub(expires, time_now());

diff --git a/lib/vconn-provider.h b/lib/vconn-provider.h
index f245e4c..1c8b86d 100644 (file)
--- a/lib/vconn-provider.h
+++ b/lib/vconn-provider.h
@@ -108,6 +108,18 @@ struct vconn_class {
      * accepted for transmission, it should return EAGAIN. */
     int (*send)(struct vconn *vconn, struct ofpbuf *msg);
 
+    /* Allows 'vconn' to perform maintenance activities, such as flushing
+     * output buffers.
+     *
+     * May be null if 'vconn' doesn't have anything to do here. */
+    void (*run)(struct vconn *vconn);
+
+    /* Arranges for the poll loop to wake up when 'vconn' needs to perform
+     * maintenance activities.
+     *
+     * May be null if 'vconn' doesn't have anything to do here. */
+    void (*run_wait)(struct vconn *vconn);
+
     /* Arranges for the poll loop to wake up when 'vconn' is ready to take an
      * action of the given 'type'. */
     void (*wait)(struct vconn *vconn, enum vconn_wait_type type);

diff --git a/lib/vconn-ssl.c b/lib/vconn-ssl.c
index 58c54f8..773090d 100644 (file)
--- a/lib/vconn-ssl.c
+++ b/lib/vconn-ssl.c
@@ -67,7 +67,6 @@ struct ssl_vconn
     SSL *ssl;
     struct ofpbuf *rxbuf;
     struct ofpbuf *txbuf;
-    struct poll_waiter *tx_waiter;
 
     /* rx_want and tx_want record the result of the last call to SSL_read()
      * and SSL_write(), respectively:
@@ -157,7 +156,6 @@ static void ssl_close(struct vconn *);
 static void ssl_clear_txbuf(struct ssl_vconn *);
 static int interpret_ssl_error(const char *function, int ret, int error,
                                int *want);
-static void ssl_tx_poll_callback(int fd, short int revents, void *vconn_);
 static DH *tmp_dh_callback(SSL *ssl, int is_export UNUSED, int keylength);
 static void log_ca_cert(const char *file_name, X509 *cert);
 
@@ -257,7 +255,6 @@ new_ssl_vconn(const char *name, int fd, enum session_type type,
     sslv->ssl = ssl;
     sslv->rxbuf = NULL;
     sslv->txbuf = NULL;
-    sslv->tx_waiter = NULL;
     sslv->rx_want = sslv->tx_want = SSL_NOTHING;
     *vconnp = &sslv->vconn;
     return 0;
@@ -441,7 +438,6 @@ static void
 ssl_close(struct vconn *vconn)
 {
     struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
-    poll_cancel(sslv->tx_waiter);
     ssl_clear_txbuf(sslv);
     ofpbuf_delete(sslv->rxbuf);
     SSL_free(sslv->ssl);
@@ -565,10 +561,6 @@ again:
     ret = SSL_read(sslv->ssl, ofpbuf_tail(rx), want_bytes);
     if (old_state != SSL_get_state(sslv->ssl)) {
         sslv->tx_want = SSL_NOTHING;
-        if (sslv->tx_waiter) {
-            poll_cancel(sslv->tx_waiter);
-            ssl_tx_poll_callback(sslv->fd, POLLIN, vconn);
-        }
     }
     sslv->rx_want = SSL_NOTHING;
 
@@ -605,16 +597,6 @@ ssl_clear_txbuf(struct ssl_vconn *sslv)
 {
     ofpbuf_delete(sslv->txbuf);
     sslv->txbuf = NULL;
-    sslv->tx_waiter = NULL;
-}
-
-static void
-ssl_register_tx_waiter(struct vconn *vconn)
-{
-    struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
-    sslv->tx_waiter = poll_fd_callback(sslv->fd,
-                                       want_to_poll_events(sslv->tx_want),
-                                       ssl_tx_poll_callback, vconn);
 }
 
 static int
@@ -647,19 +629,6 @@ ssl_do_tx(struct vconn *vconn)
     }
 }
 
-static void
-ssl_tx_poll_callback(int fd UNUSED, short int revents UNUSED, void *vconn_)
-{
-    struct vconn *vconn = vconn_;
-    struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
-    int error = ssl_do_tx(vconn);
-    if (error != EAGAIN) {
-        ssl_clear_txbuf(sslv);
-    } else {
-        ssl_register_tx_waiter(vconn);
-    }
-}
-
 static int
 ssl_send(struct vconn *vconn, struct ofpbuf *buffer)
 {
@@ -678,7 +647,6 @@ ssl_send(struct vconn *vconn, struct ofpbuf *buffer)
             return 0;
         case EAGAIN:
             leak_checker_claim(buffer);
-            ssl_register_tx_waiter(vconn);
             return 0;
         default:
             sslv->txbuf = NULL;
@@ -687,6 +655,26 @@ ssl_send(struct vconn *vconn, struct ofpbuf *buffer)
     }
 }
 
+static void
+ssl_run(struct vconn *vconn)
+{
+    struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
+
+    if (sslv->txbuf && ssl_do_tx(vconn) != EAGAIN) {
+        ssl_clear_txbuf(sslv);
+    }
+}
+
+static void
+ssl_run_wait(struct vconn *vconn)
+{
+    struct ssl_vconn *sslv = ssl_vconn_cast(vconn);
+
+    if (sslv->tx_want != SSL_NOTHING) {
+        poll_fd_wait(sslv->fd, want_to_poll_events(sslv->tx_want));
+    }
+}
+
 static void
 ssl_wait(struct vconn *vconn, enum vconn_wait_type wait)
 {
@@ -728,7 +716,8 @@ ssl_wait(struct vconn *vconn, enum vconn_wait_type wait)
             /* We have room in our tx queue. */
             poll_immediate_wake();
         } else {
-            /* The call to ssl_tx_poll_callback() will wake us up. */
+            /* vconn_run_wait() will do the right thing; don't bother with
+             * redundancy. */
         }
         break;
 
@@ -744,6 +733,8 @@ struct vconn_class ssl_vconn_class = {
     ssl_connect,                /* connect */
     ssl_recv,                   /* recv */
     ssl_send,                   /* send */
+    ssl_run,                    /* run */
+    ssl_run_wait,               /* run_wait */
     ssl_wait,                   /* wait */
 };
 \f

diff --git a/lib/vconn-stream.c b/lib/vconn-stream.c
index 0551c9e..f19f3eb 100644 (file)
--- a/lib/vconn-stream.c
+++ b/lib/vconn-stream.c
@@ -44,7 +44,6 @@ struct stream_vconn
     int fd;
     struct ofpbuf *rxbuf;
     struct ofpbuf *txbuf;
-    struct poll_waiter *tx_waiter;
     char *unlink_path;
 };
 
@@ -74,7 +73,6 @@ new_stream_vconn(const char *name, int fd, int connect_status,
     vconn_init(&s->vconn, &stream_vconn_class, connect_status, name);
     s->fd = fd;
     s->txbuf = NULL;
-    s->tx_waiter = NULL;
     s->rxbuf = NULL;
     s->unlink_path = unlink_path;
     *vconnp = &s->vconn;
@@ -92,7 +90,6 @@ static void
 stream_close(struct vconn *vconn)
 {
     struct stream_vconn *s = stream_vconn_cast(vconn);
-    poll_cancel(s->tx_waiter);
     stream_clear_txbuf(s);
     ofpbuf_delete(s->rxbuf);
     close(s->fd);
@@ -170,29 +167,6 @@ stream_clear_txbuf(struct stream_vconn *s)
 {
     ofpbuf_delete(s->txbuf);
     s->txbuf = NULL;
-    s->tx_waiter = NULL;
-}
-
-static void
-stream_do_tx(int fd UNUSED, short int revents UNUSED, void *vconn_)
-{
-    struct vconn *vconn = vconn_;
-    struct stream_vconn *s = stream_vconn_cast(vconn);
-    ssize_t n = write(s->fd, s->txbuf->data, s->txbuf->size);
-    if (n < 0) {
-        if (errno != EAGAIN) {
-            VLOG_ERR_RL(&rl, "send: %s", strerror(errno));
-            stream_clear_txbuf(s);
-            return;
-        }
-    } else if (n > 0) {
-        ofpbuf_pull(s->txbuf, n);
-        if (!s->txbuf->size) {
-            stream_clear_txbuf(s);
-            return;
-        }
-    }
-    s->tx_waiter = poll_fd_callback(s->fd, POLLOUT, stream_do_tx, vconn);
 }
 
 static int
@@ -215,13 +189,48 @@ stream_send(struct vconn *vconn, struct ofpbuf *buffer)
         if (retval > 0) {
             ofpbuf_pull(buffer, retval);
         }
-        s->tx_waiter = poll_fd_callback(s->fd, POLLOUT, stream_do_tx, vconn);
         return 0;
     } else {
         return errno;
     }
 }
 
+static void
+stream_run(struct vconn *vconn)
+{
+    struct stream_vconn *s = stream_vconn_cast(vconn);
+    ssize_t n;
+
+    if (!s->txbuf) {
+        return;
+    }
+
+    n = write(s->fd, s->txbuf->data, s->txbuf->size);
+    if (n < 0) {
+        if (errno != EAGAIN) {
+            VLOG_ERR_RL(&rl, "send: %s", strerror(errno));
+            stream_clear_txbuf(s);
+            return;
+        }
+    } else if (n > 0) {
+        ofpbuf_pull(s->txbuf, n);
+        if (!s->txbuf->size) {
+            stream_clear_txbuf(s);
+            return;
+        }
+    }
+}
+
+static void
+stream_run_wait(struct vconn *vconn)
+{
+    struct stream_vconn *s = stream_vconn_cast(vconn);
+
+    if (s->txbuf) {
+        poll_fd_wait(s->fd, POLLOUT);
+    }
+}
+
 static void
 stream_wait(struct vconn *vconn, enum vconn_wait_type wait)
 {
@@ -235,7 +244,9 @@ stream_wait(struct vconn *vconn, enum vconn_wait_type wait)
         if (!s->txbuf) {
             poll_fd_wait(s->fd, POLLOUT);
         } else {
-            /* Nothing to do: need to drain txbuf first. */
+            /* Nothing to do: need to drain txbuf first.  stream_run_wait()
+             * will arrange to wake up when there room to send data, so there's
+             * no point in calling poll_fd_wait() redundantly here. */
         }
         break;
 
@@ -255,6 +266,8 @@ static struct vconn_class stream_vconn_class = {
     stream_connect,             /* connect */
     stream_recv,                /* recv */
     stream_send,                /* send */
+    stream_run,                 /* run */
+    stream_run_wait,            /* run_wait */
     stream_wait,                /* wait */
 };
 \f

diff --git a/lib/vconn-tcp.c b/lib/vconn-tcp.c
index aac7166..e3a24f5 100644 (file)
--- a/lib/vconn-tcp.c
+++ b/lib/vconn-tcp.c
@@ -91,6 +91,8 @@ struct vconn_class tcp_vconn_class = {
     NULL,                       /* connect */
     NULL,                       /* recv */
     NULL,                       /* send */
+    NULL,                       /* run */
+    NULL,                       /* run_wait */
     NULL,                       /* wait */
 };
 \f

diff --git a/lib/vconn-unix.c b/lib/vconn-unix.c
index f637ca0..ff01022 100644 (file)
--- a/lib/vconn-unix.c
+++ b/lib/vconn-unix.c
@@ -71,6 +71,8 @@ struct vconn_class unix_vconn_class = {
     NULL,                       /* connect */
     NULL,                       /* recv */
     NULL,                       /* send */
+    NULL,                       /* run */
+    NULL,                       /* run_wait */
     NULL,                       /* wait */
 };
 \f

diff --git a/lib/vconn.c b/lib/vconn.c
index 7b9ae61..1445be4 100644 (file)
--- a/lib/vconn.c
+++ b/lib/vconn.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008, 2009 Nicira Networks.
+ * Copyright (c) 2008, 2009, 2010 Nicira Networks.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -90,7 +90,8 @@ check_vconn_classes(void)
         struct vconn_class *class = vconn_classes[i];
         assert(class->name != NULL);
         assert(class->open != NULL);
-        if (class->close || class->recv || class->send || class->wait) {
+        if (class->close || class->recv || class->send
+            || class->run || class->run_wait || class->wait) {
             assert(class->close != NULL);
             assert(class->recv != NULL);
             assert(class->send != NULL);
@@ -208,6 +209,26 @@ vconn_open(const char *name, int min_version, struct vconn **vconnp)
     return EAFNOSUPPORT;
 }
 
+/* Allows 'vconn' to perform maintenance activities, such as flushing output
+ * buffers. */
+void
+vconn_run(struct vconn *vconn)
+{
+    if (vconn->class->run) {
+        (vconn->class->run)(vconn);
+    }
+}
+
+/* Arranges for the poll loop to wake up when 'vconn' needs to perform
+ * maintenance activities. */
+void
+vconn_run_wait(struct vconn *vconn)
+{
+    if (vconn->class->run_wait) {
+        (vconn->class->run_wait)(vconn);
+    }
+}
+
 int
 vconn_open_block(const char *name, int min_version, struct vconn **vconnp)
 {
@@ -216,6 +237,8 @@ vconn_open_block(const char *name, int min_version, struct vconn **vconnp)
 
     error = vconn_open(name, min_version, &vconn);
     while (error == EAGAIN) {
+        vconn_run(vconn);
+        vconn_run_wait(vconn);
         vconn_connect_wait(vconn);
         poll_block();
         error = vconn_connect(vconn);
@@ -547,6 +570,8 @@ vconn_send_block(struct vconn *vconn, struct ofpbuf *msg)
 {
     int retval;
     while ((retval = vconn_send(vconn, msg)) == EAGAIN) {
+        vconn_run(vconn);
+        vconn_run_wait(vconn);
         vconn_send_wait(vconn);
         poll_block();
     }
@@ -559,6 +584,8 @@ vconn_recv_block(struct vconn *vconn, struct ofpbuf **msgp)
 {
     int retval;
     while ((retval = vconn_recv(vconn, msgp)) == EAGAIN) {
+        vconn_run(vconn);
+        vconn_run_wait(vconn);
         vconn_recv_wait(vconn);
         poll_block();
     }

diff --git a/lib/vconn.h b/lib/vconn.h
index 0c13744..9bd235a 100644 (file)
--- a/lib/vconn.h
+++ b/lib/vconn.h
@@ -48,6 +48,9 @@ int vconn_send(struct vconn *, struct ofpbuf *);
 int vconn_recv_xid(struct vconn *, uint32_t xid, struct ofpbuf **);
 int vconn_transact(struct vconn *, struct ofpbuf *, struct ofpbuf **);
 
+void vconn_run(struct vconn *);
+void vconn_run_wait(struct vconn *);
+
 int vconn_open_block(const char *name, int min_version, struct vconn **);
 int vconn_send_block(struct vconn *, struct ofpbuf *);
 int vconn_recv_block(struct vconn *, struct ofpbuf **);

diff --git a/tests/test-vconn.c b/tests/test-vconn.c
index 34c2930..87e35c4 100644 (file)
--- a/tests/test-vconn.c
+++ b/tests/test-vconn.c
@@ -143,6 +143,7 @@ test_refuse_connection(const char *type, int expected_error)
     fpv_create(type, &fpv);
     assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
     fpv_close(&fpv);
+    vconn_run(vconn);
     assert(vconn_connect(vconn) == expected_error);
     vconn_close(vconn);
     fpv_destroy(&fpv);
@@ -159,6 +160,7 @@ test_accept_then_close(const char *type, int expected_error)
 
     fpv_create(type, &fpv);
     assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
+    vconn_run(vconn);
     close(fpv_accept(&fpv));
     fpv_close(&fpv);
     assert(vconn_connect(vconn) == expected_error);
@@ -178,6 +180,7 @@ test_read_hello(const char *type, int expected_error)
 
     fpv_create(type, &fpv);
     assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
+    vconn_run(vconn);
     fd = fpv_accept(&fpv);
     fpv_destroy(&fpv);
     assert(!set_nonblocking(fd));
@@ -195,7 +198,9 @@ test_read_hello(const char *type, int expected_error)
            assert(errno == EAGAIN);
        }
 
+       vconn_run(vconn);
        assert(vconn_connect(vconn) == EAGAIN);
+       vconn_run_wait(vconn);
        vconn_connect_wait(vconn);
        poll_fd_wait(fd, POLLIN);
        poll_block();
@@ -221,6 +226,7 @@ test_send_hello(const char *type, const void *out, size_t out_size,
 
     fpv_create(type, &fpv);
     assert(!vconn_open(fpv.vconn_name, OFP_VERSION, &vconn));
+    vconn_run(vconn);
     fd = fpv_accept(&fpv);
     fpv_destroy(&fpv);
 
@@ -243,6 +249,7 @@ test_send_hello(const char *type, const void *out, size_t out_size,
            }
        }
 
+       vconn_run(vconn);
        if (!connected) {
            int error = vconn_connect(vconn);
            if (error == expect_connect_error) {
@@ -262,6 +269,7 @@ test_send_hello(const char *type, const void *out, size_t out_size,
            break;
        }
 
+       vconn_run_wait(vconn);
        if (!connected) {
            vconn_connect_wait(vconn);
        }