[ovs-dev] [PATCH 3/3] ovsdb-server: Sync remotes with the monitor.

Gurucharan Shetty shettyg at nicira.com
Mon Jun 10 08:10:30 UTC 2013


commit b421d2af0ab (ovsdb-server: Add commands for adding and removing remotes)
made it possible to make ovsdb-server connect to OVS managers only after
ovs-vswitchd has completed its initial configuration. But this results in an
undesirable effect. Whenever ovsdb-server crashes, the monitor restarts its.
But ovsdb-server can no longer connect to the manager because the remotes were
added during runtime and that information is lost during the crash.

This commit lets the main ovsdb-server to sync the data that should not get
lost because of a crash. When the monitor restarts ovsdb-server, the data
is still available.

Signed-off-by: Gurucharan Shetty <gshetty at nicira.com>
---
 lib/rxbuf.c          |    7 +++
 lib/rxbuf.h          |    1 +
 ovsdb/automake.mk    |    2 +
 ovsdb/ovsdb-server.c |   75 +++++++++++++++++++++++++++-
 ovsdb/sync.c         |  135 ++++++++++++++++++++++++++++++++++++++++++++++++++
 ovsdb/sync.h         |   39 +++++++++++++++
 6 files changed, 258 insertions(+), 1 deletion(-)
 create mode 100644 ovsdb/sync.c
 create mode 100644 ovsdb/sync.h

diff --git a/lib/rxbuf.c b/lib/rxbuf.c
index 567f3a1..d988a63 100644
--- a/lib/rxbuf.c
+++ b/lib/rxbuf.c
@@ -26,6 +26,13 @@ rxbuf_init(struct rxbuf *rx)
 }
 
 void
+rxbuf_unint(struct rxbuf *rx)
+{
+    ofpbuf_uninit(&rx->header);
+    ofpbuf_uninit(&rx->payload);
+}
+
+void
 rxbuf_clear(struct rxbuf *rx)
 {
     ofpbuf_clear(&rx->header);
diff --git a/lib/rxbuf.h b/lib/rxbuf.h
index a6af7e0..94bcc42 100644
--- a/lib/rxbuf.h
+++ b/lib/rxbuf.h
@@ -32,6 +32,7 @@ struct rxbuf {
 };
 
 void rxbuf_init(struct rxbuf *);
+void rxbuf_uninit(struct rxbuf *);
 void rxbuf_clear(struct rxbuf *);
 int rxbuf_run(struct rxbuf *, int sock, size_t header_len);
 
diff --git a/ovsdb/automake.mk b/ovsdb/automake.mk
index d2e3f9a..f3fb990 100644
--- a/ovsdb/automake.mk
+++ b/ovsdb/automake.mk
@@ -23,6 +23,8 @@ ovsdb_libovsdb_a_SOURCES = \
 	ovsdb/row.h \
 	ovsdb/server.c \
 	ovsdb/server.h \
+	ovsdb/sync.c \
+	ovsdb/sync.h \
 	ovsdb/table.c \
 	ovsdb/table.h \
 	ovsdb/trigger.c \
diff --git a/ovsdb/ovsdb-server.c b/ovsdb/ovsdb-server.c
index 6311c6b..763fe12 100644
--- a/ovsdb/ovsdb-server.c
+++ b/ovsdb/ovsdb-server.c
@@ -47,6 +47,7 @@
 #include "stream.h"
 #include "stress.h"
 #include "sset.h"
+#include "sync.h"
 #include "table.h"
 #include "timeval.h"
 #include "transaction.h"
@@ -99,6 +100,13 @@ static void update_remote_status(const struct ovsdb_jsonrpc_server *jsonrpc,
                                  const struct sset *remotes,
                                  struct db dbs[], size_t n_dbs);
 
+/* Maintain the remotes added to a monitored process. */
+static struct sset remotes_sync;
+static void remotes_init(struct sset *remotes);
+
+static monitor_callback handle_monitor_callback;
+static sync_data_callback update_sync_data;
+
 int
 main(int argc, char *argv[])
 {
@@ -112,6 +120,7 @@ main(int argc, char *argv[])
     int retval;
     long long int status_timer = LLONG_MIN;
     struct add_remote_aux add_remote_aux;
+    const char *remote_name;
 
     struct db *dbs;
     int n_dbs;
@@ -125,7 +134,16 @@ main(int argc, char *argv[])
 
     parse_options(&argc, &argv, &remotes, &unixctl_path, &run_command);
 
-    daemonize_start(NULL);
+    if (daemon_get_monitor()) {
+        sync_init();
+    }
+    remotes_init(&remotes);
+
+    daemonize_start(handle_monitor_callback);
+
+    SSET_FOR_EACH (remote_name, &remotes_sync) {
+        sset_add(&remotes, remote_name);
+    }
 
     n_dbs = MAX(1, argc);
     dbs = xcalloc(n_dbs + 1, sizeof *dbs);
@@ -273,6 +291,49 @@ main(int argc, char *argv[])
     return 0;
 }
 
+static void
+remotes_init(struct sset *remotes)
+{
+    const char *remote_name;
+    sset_init(&remotes_sync);
+    SSET_FOR_EACH (remote_name, remotes) {
+        sset_add(&remotes_sync, remote_name);
+    }
+    sset_clear(remotes);
+}
+
+/* Callback from the monitor. */
+static void
+handle_monitor_callback(void)
+{
+    sync_receive_monitor(update_sync_data);
+}
+
+/* When ovsdb-sever is monitored by a 'monitor' and data that should remain
+ * constant across monitor restarts is changed in the monitored process,
+ * change the corresponding data based on 'request_type' in the monitor. */
+static void
+update_sync_data(enum sync_request_type request_type, struct ofpbuf *reply)
+{
+    char *data;
+    switch(request_type) {
+    case SYNC_REMOTE_ADD:
+        data = xmalloc(reply->size+1);
+        ovs_strlcpy(data, reply->data, reply->size+1);
+        sset_add(&remotes_sync, data);
+        free(data);
+        break;
+    case SYNC_REMOTE_REMOVE:
+        data = xmalloc(reply->size+1);
+        ovs_strlcpy(data, reply->data, reply->size+1);
+        sset_find_and_delete(&remotes_sync, data);
+        free(data);
+        break;
+    default:
+        VLOG_ERR("Uknown sync request_type.");
+    }
+}
+
 static const struct db *
 find_db(const struct db dbs[], size_t n_dbs, const char *db_name)
 {
@@ -913,12 +974,18 @@ ovsdb_server_add_remote(struct unixctl_conn *conn, int argc OVS_UNUSED,
     const struct ovsdb_table *table;
     const struct db *db;
     char *retval;
+    enum sync_request_type request_type = SYNC_REMOTE_ADD;
 
     retval = (strncmp("db:", remote, 3)
               ? NULL
               : parse_db_column(aux->dbs, aux->n_dbs, remote,
                                 &db, &table, &column));
     if (!retval) {
+        if (daemon_get_monitor() &&
+                !!sync_send_monitor(remote, request_type)) {
+            unixctl_command_reply_error(conn, "system error.");
+            return;
+        }
         sset_add(aux->remotes, remote);
         unixctl_command_reply(conn, NULL);
     } else {
@@ -935,9 +1002,15 @@ ovsdb_server_remove_remote(struct unixctl_conn *conn, int argc OVS_UNUSED,
 {
     struct sset *remotes = remotes_;
     struct sset_node *node;
+    enum sync_request_type request_type = SYNC_REMOTE_REMOVE;
 
     node = sset_find(remotes, argv[1]);
     if (node) {
+        if (daemon_get_monitor() &&
+                !!sync_send_monitor(argv[1], request_type)) {
+            unixctl_command_reply_error(conn, "system error.");
+            return;
+        }
         sset_delete(remotes, node);
         unixctl_command_reply(conn, NULL);
     } else {
diff --git a/ovsdb/sync.c b/ovsdb/sync.c
new file mode 100644
index 0000000..d108ecd
--- /dev/null
+++ b/ovsdb/sync.c
@@ -0,0 +1,135 @@
+/* Copyright (c) 2013 Nicira, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <config.h>
+
+#include "sync.h"
+
+#include <errno.h>
+#include <poll.h>
+#include <stddef.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/uio.h>
+#include <sys/wait.h>
+#include <unistd.h>
+
+#include "poll-loop.h"
+#include "rxbuf.h"
+#include "socket-util.h"
+#include "util.h"
+#include "vlog.h"
+
+VLOG_DEFINE_THIS_MODULE(sync);
+
+static int sync_fds[2];
+
+void
+sync_init(void)
+{
+    xsocketpair(AF_UNIX, SOCK_STREAM, 0, sync_fds);
+    xset_nonblocking(sync_fds[0]);
+    xset_nonblocking(sync_fds[1]);
+}
+
+/* Runs in a monitor in an infinite loop waiting to receive requests from
+ * a monitored process to sync data. The function returns in case the
+ * monitored process dies. */
+void
+sync_receive_monitor(sync_data_callback *cb)
+{
+    int error;
+    struct sync_send_request request;
+    struct rxbuf rx;
+
+    if (sync_fds[1]) {
+        close(sync_fds[1]);
+        sync_fds[1] = -1;
+    }
+
+    rxbuf_init(&rx);
+    for(;;) {
+        error = rxbuf_run(&rx, sync_fds[0], sizeof(struct sync_send_request));
+        if (!error) {
+            request = *(struct sync_send_request *) rx.header.data;
+            cb(request.request_type, &rx.payload);
+            rxbuf_clear(&rx);
+        } else if (error == EOF && !rx.header.size) {
+            /* The child closed the IPC socket. Exit cleanly. */
+            break;
+        } else if (error != EAGAIN) {
+            VLOG_ERR("RPC receive failed (%s)", ovs_retval_to_string(error));
+            break;
+        }
+        poll_fd_wait(sync_fds[0], POLLIN|POLLRDHUP|POLLERR);
+        poll_block();
+    }
+    rxbuf_unint(&rx);
+    close(sync_fds[0]);
+    sync_init();
+}
+
+/* Called by a monitored process with customized 'data' and a 'request_type'
+ * that is understood by the monitor process. */
+int
+sync_send_monitor(const char *data, enum sync_request_type request_type)
+{
+    struct sync_send_request rq;
+    int error;
+    size_t sent = 0;
+    struct pollfd pfd;
+
+    if (sync_fds[0]) {
+        sync_fds[0] = -1;
+        close(sync_fds[0]);
+    }
+
+    if (strlen(data) > 0) {
+        struct iovec iov[2];
+
+        memset(iov, 0, sizeof(iov));
+        memset(&rq, 0, sizeof(rq));
+
+        rq.request_len = strlen(data);
+        rq.request_type = request_type;
+        iov[0].iov_base = (void *) &rq;
+        iov[0].iov_len = sizeof(rq);
+
+        iov[1].iov_base = (void *) data;
+        iov[1].iov_len = strlen(data);
+
+        for(;;) {
+            error = send_iovec_and_fds_fully(sync_fds[1], iov, 2, NULL, 0,
+                                            sent, &sent);
+            if (error != EAGAIN) {
+                return error;
+            }
+
+            /* Wait for 'client_sock' to become ready before trying again. */
+            pfd.fd = sync_fds[1];
+            pfd.events = POLLOUT;
+            do {
+                error = poll(&pfd, 1, -1) < 0 ? errno : 0;
+            } while (error == EINTR);
+            if (error) {
+                VLOG_ERR("poll failed (%s)", strerror(error));
+                return error;
+            }
+        }
+    }
+    return 0;
+}
diff --git a/ovsdb/sync.h b/ovsdb/sync.h
new file mode 100644
index 0000000..0b05eeb
--- /dev/null
+++ b/ovsdb/sync.h
@@ -0,0 +1,39 @@
+/* Copyright (c) 2013 Nicira, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SYNC_H
+#define SYNC_H 1
+
+#include <stdbool.h>
+#include <stddef.h>
+
+#include "ofpbuf.h"
+
+enum sync_request_type {
+    SYNC_REMOTE_ADD,        /* Add remote manager. */
+    SYNC_REMOTE_REMOVE      /* Remove remote manager. */
+};
+
+struct sync_send_request {
+    size_t request_len;                   /* Length of the payload in bytes. */
+    enum sync_request_type request_type;  /* Type of payload. */
+};
+
+void sync_init(void);
+int sync_send_monitor(const char *data, enum sync_request_type request_type);
+typedef void sync_data_callback(enum sync_request_type, struct ofpbuf *);
+void sync_receive_monitor(sync_data_callback *cb);
+
+#endif /* sync.h */
-- 
1.7.9.5




More information about the dev mailing list