Setting tag sliver-openvswitch-2.2.90-1
[sliver-openvswitch.git] / ovsdb / ovsdb-client.c
index e8afdd6..2942953 100644 (file)
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009, 2010, 2011 Nicira Networks.
+ * Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014 Nicira, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
 
 #include <config.h>
 
-#include <assert.h>
+#include <ctype.h>
 #include <errno.h>
 #include <getopt.h>
 #include <limits.h>
 #include "column.h"
 #include "compiler.h"
 #include "daemon.h"
+#include "dirs.h"
 #include "dynamic-string.h"
+#include "fatal-signal.h"
 #include "json.h"
 #include "jsonrpc.h"
 #include "lib/table.h"
 #include "ovsdb.h"
 #include "ovsdb-data.h"
 #include "ovsdb-error.h"
+#include "poll-loop.h"
 #include "sort.h"
+#include "svec.h"
 #include "stream.h"
 #include "stream-ssl.h"
 #include "table.h"
 #include "timeval.h"
+#include "unixctl.h"
 #include "util.h"
 #include "vlog.h"
 
 VLOG_DEFINE_THIS_MODULE(ovsdb_client);
 
+enum args_needed {
+    NEED_NONE,            /* No JSON-RPC connection or database name needed. */
+    NEED_RPC,             /* JSON-RPC connection needed. */
+    NEED_DATABASE         /* JSON-RPC connection and database name needed. */
+};
+
+struct ovsdb_client_command {
+    const char *name;
+    enum args_needed need;
+    int min_args;
+    int max_args;
+    void (*handler)(struct jsonrpc *rpc, const char *database,
+                    int argc, char *argv[]);
+};
+
+/* --timestamp: Print a timestamp before each update on "monitor" command? */
+static bool timestamp;
+
 /* Format for table output. */
 static struct table_style table_style = TABLE_STYLE_DEFAULT;
 
-static const struct command all_commands[];
+static const struct ovsdb_client_command *get_all_commands(void);
 
 static void usage(void) NO_RETURN;
 static void parse_options(int argc, char *argv[]);
+static struct jsonrpc *open_jsonrpc(const char *server);
+static void fetch_dbs(struct jsonrpc *, struct svec *dbs);
 
 int
 main(int argc, char *argv[])
 {
+    const struct ovsdb_client_command *command;
+    const char *database;
+    struct jsonrpc *rpc;
+
     proctitle_init(argc, argv);
     set_program_name(argv[0]);
     parse_options(argc, argv);
-    signal(SIGPIPE, SIG_IGN);
-    run_command(argc - optind, argv + optind, all_commands);
+    fatal_ignore_sigpipe();
+
+    if (optind >= argc) {
+        ovs_fatal(0, "missing command name; use --help for help");
+    }
+
+    for (command = get_all_commands(); ; command++) {
+        if (!command->name) {
+            VLOG_FATAL("unknown command '%s'; use --help for help",
+                       argv[optind]);
+        } else if (!strcmp(command->name, argv[optind])) {
+            break;
+        }
+    }
+    optind++;
+
+    if (command->need != NEED_NONE) {
+        if (argc - optind > command->min_args
+            && (isalpha((unsigned char) argv[optind][0])
+                && strchr(argv[optind], ':'))) {
+            rpc = open_jsonrpc(argv[optind++]);
+        } else {
+            char *sock = xasprintf("unix:%s/db.sock", ovs_rundir());
+            rpc = open_jsonrpc(sock);
+            free(sock);
+        }
+    } else {
+        rpc = NULL;
+    }
+
+    if (command->need == NEED_DATABASE) {
+        struct svec dbs;
+
+        svec_init(&dbs);
+        fetch_dbs(rpc, &dbs);
+        if (argc - optind > command->min_args
+            && svec_contains(&dbs, argv[optind])) {
+            database = argv[optind++];
+        } else if (dbs.n == 1) {
+            database = xstrdup(dbs.names[0]);
+        } else if (svec_contains(&dbs, "Open_vSwitch")) {
+            database = "Open_vSwitch";
+        } else {
+            ovs_fatal(0, "no default database for `%s' command, please "
+                      "specify a database name", command->name);
+        }
+        svec_destroy(&dbs);
+    } else {
+        database = NULL;
+    }
+
+    if (argc - optind < command->min_args ||
+        argc - optind > command->max_args) {
+        VLOG_FATAL("invalid syntax for '%s' (use --help for help)",
+                    command->name);
+    }
+
+    command->handler(rpc, database, argc - optind, argv + optind);
+
+    jsonrpc_close(rpc);
+
+    if (ferror(stdout)) {
+        VLOG_FATAL("write to stdout failed");
+    }
+    if (ferror(stderr)) {
+        VLOG_FATAL("write to stderr failed");
+    }
+
     return 0;
 }
 
@@ -70,20 +165,23 @@ parse_options(int argc, char *argv[])
 {
     enum {
         OPT_BOOTSTRAP_CA_CERT = UCHAR_MAX + 1,
+        OPT_TIMESTAMP,
+        VLOG_OPTION_ENUMS,
         DAEMON_OPTION_ENUMS,
         TABLE_OPTION_ENUMS
     };
-    static struct option long_options[] = {
-        {"verbose", optional_argument, 0, 'v'},
-        {"help", no_argument, 0, 'h'},
-        {"version", no_argument, 0, 'V'},
+    static const struct option long_options[] = {
+        {"help", no_argument, NULL, 'h'},
+        {"version", no_argument, NULL, 'V'},
+        {"timestamp", no_argument, NULL, OPT_TIMESTAMP},
+        VLOG_LONG_OPTIONS,
         DAEMON_LONG_OPTIONS,
 #ifdef HAVE_OPENSSL
-        {"bootstrap-ca-cert", required_argument, 0, OPT_BOOTSTRAP_CA_CERT},
+        {"bootstrap-ca-cert", required_argument, NULL, OPT_BOOTSTRAP_CA_CERT},
         STREAM_SSL_LONG_OPTIONS,
 #endif
         TABLE_LONG_OPTIONS,
-        {0, 0, 0, 0},
+        {NULL, 0, NULL, 0},
     };
     char *short_options = long_options_to_short_options(long_options);
 
@@ -100,23 +198,22 @@ parse_options(int argc, char *argv[])
             usage();
 
         case 'V':
-            OVS_PRINT_VERSION(0, 0);
+            ovs_print_version(0, 0);
             exit(EXIT_SUCCESS);
 
-        case 'v':
-            vlog_set_verbosity(optarg);
-            break;
-
+        VLOG_OPTION_HANDLERS
         DAEMON_OPTION_HANDLERS
-
         TABLE_OPTION_HANDLERS(&table_style)
-
         STREAM_SSL_OPTION_HANDLERS
 
         case OPT_BOOTSTRAP_CA_CERT:
             stream_ssl_set_ca_cert_file(optarg, true);
             break;
 
+        case OPT_TIMESTAMP:
+            timestamp = true;
+            break;
+
         case '?':
             exit(EXIT_FAILURE);
 
@@ -137,34 +234,40 @@ usage(void)
     printf("%s: Open vSwitch database JSON-RPC client\n"
            "usage: %s [OPTIONS] COMMAND [ARG...]\n"
            "\nValid commands are:\n"
-           "\n  list-dbs SERVER\n"
+           "\n  list-dbs [SERVER]\n"
            "    list databases available on SERVER\n"
-           "\n  get-schema SERVER DATABASE\n"
+           "\n  get-schema [SERVER] [DATABASE]\n"
            "    retrieve schema for DATABASE from SERVER\n"
-           "\n  get-schema-version SERVER DATABASE\n"
+           "\n  get-schema-version [SERVER] [DATABASE]\n"
            "    retrieve schema for DATABASE from SERVER and report only its\n"
            "    version number on stdout\n"
-           "\n  list-tables SERVER DATABASE\n"
+           "\n  list-tables [SERVER] [DATABASE]\n"
            "    list tables for DATABASE on SERVER\n"
-           "\n  list-columns SERVER DATABASE [TABLE]\n"
+           "\n  list-columns [SERVER] [DATABASE] [TABLE]\n"
            "    list columns in TABLE (or all tables) in DATABASE on SERVER\n"
-           "\n  transact SERVER TRANSACTION\n"
+           "\n  transact [SERVER] TRANSACTION\n"
            "    run TRANSACTION (a JSON array of operations) on SERVER\n"
            "    and print the results as JSON on stdout\n"
-           "\n  monitor SERVER DATABASE TABLE [COLUMN,...]...\n"
+           "\n  monitor [SERVER] [DATABASE] TABLE [COLUMN,...]...\n"
            "    monitor contents of COLUMNs in TABLE in DATABASE on SERVER.\n"
            "    COLUMNs may include !initial, !insert, !delete, !modify\n"
            "    to avoid seeing the specified kinds of changes.\n"
-           "\n  dump SERVER DATABASE\n"
-           "    dump contents of DATABASE on SERVER to stdout\n",
-           program_name, program_name);
+           "\n  monitor [SERVER] [DATABASE] ALL\n"
+           "    monitor all changes to all columns in all tables\n"
+           "    in DATBASE on SERVER.\n"
+           "\n  dump [SERVER] [DATABASE]\n"
+           "    dump contents of DATABASE on SERVER to stdout\n"
+           "\nThe default SERVER is unix:%s/db.sock.\n"
+           "The default DATABASE is Open_vSwitch.\n",
+           program_name, program_name, ovs_rundir());
     stream_usage("SERVER", true, true, true);
     printf("\nOutput formatting options:\n"
            "  -f, --format=FORMAT         set output formatting to FORMAT\n"
            "                              (\"table\", \"html\", \"csv\", "
            "or \"json\")\n"
            "  --no-headings               omit table heading row\n"
-           "  --pretty                    pretty-print JSON in output");
+           "  --pretty                    pretty-print JSON in output\n"
+           "  --timestamp                 timestamp \"monitor\" output");
     daemon_usage();
     vlog_usage();
     printf("\nOther options:\n"
@@ -173,6 +276,21 @@ usage(void)
     exit(EXIT_SUCCESS);
 }
 \f
+static void
+check_txn(int error, struct jsonrpc_msg **reply_)
+{
+    struct jsonrpc_msg *reply = *reply_;
+
+    if (error) {
+        ovs_fatal(error, "transaction failed");
+    }
+
+    if (reply->error) {
+        ovs_fatal(error, "transaction returned error: %s",
+                  json_to_string(reply->error, table_style.json_flags));
+    }
+}
+
 static struct json *
 parse_json(const char *s)
 {
@@ -189,11 +307,12 @@ open_jsonrpc(const char *server)
     struct stream *stream;
     int error;
 
-    error = stream_open_block(jsonrpc_stream_open(server, &stream), &stream);
+    error = stream_open_block(jsonrpc_stream_open(server, &stream,
+                              DSCP_DEFAULT), &stream);
     if (error == EAFNOSUPPORT) {
         struct pstream *pstream;
 
-        error = jsonrpc_pstream_open(server, &pstream);
+        error = jsonrpc_pstream_open(server, &pstream, DSCP_DEFAULT);
         if (error) {
             ovs_fatal(error, "failed to connect or listen to \"%s\"", server);
         }
@@ -236,56 +355,32 @@ check_ovsdb_error(struct ovsdb_error *error)
 }
 
 static struct ovsdb_schema *
-fetch_schema_from_rpc(struct jsonrpc *rpc, const char *database)
+fetch_schema(struct jsonrpc *rpc, const char *database)
 {
     struct jsonrpc_msg *request, *reply;
     struct ovsdb_schema *schema;
-    int error;
 
     request = jsonrpc_create_request("get_schema",
                                      json_array_create_1(
                                          json_string_create(database)),
                                      NULL);
-    error = jsonrpc_transact_block(rpc, request, &reply);
-    if (error) {
-        ovs_fatal(error, "transaction failed");
-    }
+    check_txn(jsonrpc_transact_block(rpc, request, &reply), &reply);
     check_ovsdb_error(ovsdb_schema_from_json(reply->result, &schema));
     jsonrpc_msg_destroy(reply);
 
     return schema;
 }
 
-static struct ovsdb_schema *
-fetch_schema(const char *server, const char *database)
-{
-    struct ovsdb_schema *schema;
-    struct jsonrpc *rpc;
-
-    rpc = open_jsonrpc(server);
-    schema = fetch_schema_from_rpc(rpc, database);
-    jsonrpc_close(rpc);
-
-    return schema;
-}
-\f
-\f
 static void
-do_list_dbs(int argc OVS_UNUSED, char *argv[])
+fetch_dbs(struct jsonrpc *rpc, struct svec *dbs)
 {
     struct jsonrpc_msg *request, *reply;
-    struct jsonrpc *rpc;
-    int error;
     size_t i;
 
-    rpc = open_jsonrpc(argv[1]);
     request = jsonrpc_create_request("list_dbs", json_array_create_empty(),
                                      NULL);
-    error = jsonrpc_transact_block(rpc, request, &reply);
-    if (error) {
-        ovs_fatal(error, "transaction failed");
-    }
 
+    check_txn(jsonrpc_transact_block(rpc, request, &reply), &reply);
     if (reply->result->type != JSON_ARRAY) {
         ovs_fatal(0, "list_dbs response is not array");
     }
@@ -294,37 +389,57 @@ do_list_dbs(int argc OVS_UNUSED, char *argv[])
         const struct json *name = reply->result->u.array.elems[i];
 
         if (name->type != JSON_STRING) {
-            ovs_fatal(0, "list_dbs response %zu is not string", i);
+            ovs_fatal(0, "list_dbs response %"PRIuSIZE" is not string", i);
         }
-        puts(name->u.string);
+        svec_add(dbs, name->u.string);
     }
     jsonrpc_msg_destroy(reply);
+    svec_sort(dbs);
+}
+\f
+static void
+do_list_dbs(struct jsonrpc *rpc, const char *database OVS_UNUSED,
+            int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
+{
+    const char *db_name;
+    struct svec dbs;
+    size_t i;
+
+    svec_init(&dbs);
+    fetch_dbs(rpc, &dbs);
+    SVEC_FOR_EACH (i, db_name, &dbs) {
+        puts(db_name);
+    }
+    svec_destroy(&dbs);
 }
 
 static void
-do_get_schema(int argc OVS_UNUSED, char *argv[])
+do_get_schema(struct jsonrpc *rpc, const char *database,
+              int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
 {
-    struct ovsdb_schema *schema = fetch_schema(argv[1], argv[2]);
+    struct ovsdb_schema *schema = fetch_schema(rpc, database);
     print_and_free_json(ovsdb_schema_to_json(schema));
     ovsdb_schema_destroy(schema);
 }
 
 static void
-do_get_schema_version(int argc OVS_UNUSED, char *argv[])
+do_get_schema_version(struct jsonrpc *rpc, const char *database,
+                      int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
 {
-    struct ovsdb_schema *schema = fetch_schema(argv[1], argv[2]);
+    struct ovsdb_schema *schema = fetch_schema(rpc, database);
     puts(schema->version);
     ovsdb_schema_destroy(schema);
 }
 
 static void
-do_list_tables(int argc OVS_UNUSED, char *argv[])
+do_list_tables(struct jsonrpc *rpc, const char *database,
+               int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
 {
     struct ovsdb_schema *schema;
     struct shash_node *node;
     struct table t;
 
-    schema = fetch_schema(argv[1], argv[2]);
+    schema = fetch_schema(rpc, database);
     table_init(&t);
     table_add_column(&t, "Table");
     SHASH_FOR_EACH (node, &schema->tables) {
@@ -338,14 +453,15 @@ do_list_tables(int argc OVS_UNUSED, char *argv[])
 }
 
 static void
-do_list_columns(int argc OVS_UNUSED, char *argv[])
+do_list_columns(struct jsonrpc *rpc, const char *database,
+                int argc OVS_UNUSED, char *argv[])
 {
-    const char *table_name = argv[3];
+    const char *table_name = argv[0];
     struct ovsdb_schema *schema;
     struct shash_node *table_node;
     struct table t;
 
-    schema = fetch_schema(argv[1], argv[2]);
+    schema = fetch_schema(rpc, database);
     table_init(&t);
     if (!table_name) {
         table_add_column(&t, "Table");
@@ -375,30 +491,27 @@ do_list_columns(int argc OVS_UNUSED, char *argv[])
 }
 
 static void
-do_transact(int argc OVS_UNUSED, char *argv[])
+do_transact(struct jsonrpc *rpc, const char *database OVS_UNUSED,
+            int argc OVS_UNUSED, char *argv[])
 {
     struct jsonrpc_msg *request, *reply;
     struct json *transaction;
-    struct jsonrpc *rpc;
-    int error;
 
-    transaction = parse_json(argv[2]);
+    transaction = parse_json(argv[0]);
 
-    rpc = open_jsonrpc(argv[1]);
     request = jsonrpc_create_request("transact", transaction, NULL);
-    error = jsonrpc_transact_block(rpc, request, &reply);
-    if (error) {
-        ovs_fatal(error, "transaction failed");
-    }
-    if (reply->error) {
-        ovs_fatal(error, "transaction returned error: %s",
-                  json_to_string(reply->error, table_style.json_flags));
-    }
+    check_txn(jsonrpc_transact_block(rpc, request, &reply), &reply);
     print_json(reply->result);
     putchar('\n');
     jsonrpc_msg_destroy(reply);
-    jsonrpc_close(rpc);
 }
+\f
+/* "monitor" command. */
+
+struct monitored_table {
+    struct ovsdb_table_schema *table;
+    struct ovsdb_column_set columns;
+};
 
 static void
 monitor_print_row(struct json *row, const char *type, const char *uuid,
@@ -429,30 +542,25 @@ monitor_print_row(struct json *row, const char *type, const char *uuid,
 }
 
 static void
-monitor_print(struct json *table_updates,
-              const struct ovsdb_table_schema *table,
-              const struct ovsdb_column_set *columns, bool initial)
+monitor_print_table(struct json *table_update,
+                    const struct monitored_table *mt, char *caption,
+                    bool initial)
 {
-    struct json *table_update;
+    const struct ovsdb_table_schema *table = mt->table;
+    const struct ovsdb_column_set *columns = &mt->columns;
     struct shash_node *node;
     struct table t;
     size_t i;
 
-    table_init(&t);
-
-    if (table_updates->type != JSON_OBJECT) {
-        ovs_error(0, "<table-updates> is not object");
-        return;
-    }
-    table_update = shash_find_data(json_object(table_updates), table->name);
-    if (!table_update) {
-        return;
-    }
     if (table_update->type != JSON_OBJECT) {
-        ovs_error(0, "<table-update> is not object");
+        ovs_error(0, "<table-update> for table %s is not object", table->name);
         return;
     }
 
+    table_init(&t);
+    table_set_timestamp(&t, timestamp);
+    table_set_caption(&t, caption);
+
     table_add_column(&t, "row");
     table_add_column(&t, "action");
     for (i = 0; i < columns->n_columns; i++) {
@@ -483,6 +591,30 @@ monitor_print(struct json *table_updates,
     table_destroy(&t);
 }
 
+static void
+monitor_print(struct json *table_updates,
+              const struct monitored_table *mts, size_t n_mts,
+              bool initial)
+{
+    size_t i;
+
+    if (table_updates->type != JSON_OBJECT) {
+        ovs_error(0, "<table-updates> is not object");
+        return;
+    }
+
+    for (i = 0; i < n_mts; i++) {
+        const struct monitored_table *mt = &mts[i];
+        struct json *table_update = shash_find_data(json_object(table_updates),
+                                                    mt->table->name);
+        if (table_update) {
+            monitor_print_table(table_update, mt,
+                                n_mts > 1 ? xstrdup(mt->table->name) : NULL,
+                                initial);
+        }
+    }
+}
+
 static void
 add_column(const char *server, const struct ovsdb_column *column,
            struct ovsdb_column_set *columns, struct json *columns_json)
@@ -548,7 +680,7 @@ parse_monitor_columns(char *arg, const char *server, const char *database,
         }
         free(nodes);
 
-        add_column(server, ovsdb_table_schema_get_column(table,"_version"),
+        add_column(server, ovsdb_table_schema_get_column(table, "_version"),
                    columns, columns_json);
     }
 
@@ -565,92 +697,202 @@ parse_monitor_columns(char *arg, const char *server, const char *database,
 }
 
 static void
-do_monitor(int argc, char *argv[])
+ovsdb_client_exit(struct unixctl_conn *conn, int argc OVS_UNUSED,
+                  const char *argv[] OVS_UNUSED, void *exiting_)
 {
-    const char *server = argv[1];
-    const char *database = argv[2];
-    const char *table_name = argv[3];
-    struct ovsdb_column_set columns = OVSDB_COLUMN_SET_INITIALIZER;
-    struct ovsdb_table_schema *table;
-    struct ovsdb_schema *schema;
-    struct jsonrpc_msg *request;
-    struct jsonrpc *rpc;
-    struct json *monitor, *monitor_request_array,
-        *monitor_requests, *request_id;
+    bool *exiting = exiting_;
+    *exiting = true;
+    unixctl_command_reply(conn, NULL);
+}
+
+static void
+ovsdb_client_block(struct unixctl_conn *conn, int argc OVS_UNUSED,
+                   const char *argv[] OVS_UNUSED, void *blocked_)
+{
+    bool *blocked = blocked_;
+
+    if (!*blocked) {
+        *blocked = true;
+        unixctl_command_reply(conn, NULL);
+    } else {
+        unixctl_command_reply(conn, "already blocking");
+    }
+}
+
+static void
+ovsdb_client_unblock(struct unixctl_conn *conn, int argc OVS_UNUSED,
+                     const char *argv[] OVS_UNUSED, void *blocked_)
+{
+    bool *blocked = blocked_;
+
+    if (*blocked) {
+        *blocked = false;
+        unixctl_command_reply(conn, NULL);
+    } else {
+        unixctl_command_reply(conn, "already unblocked");
+    }
+}
 
-    rpc = open_jsonrpc(server);
+static void
+add_monitored_table(int argc, char *argv[],
+                    const char *server, const char *database,
+                    struct ovsdb_table_schema *table,
+                    struct json *monitor_requests,
+                    struct monitored_table **mts,
+                    size_t *n_mts, size_t *allocated_mts)
+{
+    struct json *monitor_request_array;
+    struct monitored_table *mt;
 
-    schema = fetch_schema_from_rpc(rpc, database);
-    table = shash_find_data(&schema->tables, table_name);
-    if (!table) {
-        ovs_fatal(0, "%s: %s does not have a table named \"%s\"",
-                  server, database, table_name);
+    if (*n_mts >= *allocated_mts) {
+        *mts = x2nrealloc(*mts, allocated_mts, sizeof **mts);
     }
+    mt = &(*mts)[(*n_mts)++];
+    mt->table = table;
+    ovsdb_column_set_init(&mt->columns);
 
     monitor_request_array = json_array_create_empty();
-    if (argc > 4) {
+    if (argc > 1) {
         int i;
 
-        for (i = 4; i < argc; i++) {
+        for (i = 1; i < argc; i++) {
             json_array_add(
                 monitor_request_array,
                 parse_monitor_columns(argv[i], server, database, table,
-                                      &columns));
+                                      &mt->columns));
         }
     } else {
-        /* Allocate a writable empty string since parse_monitor_columns() is
-         * going to strtok() it and that's risky with literal "". */
+        /* Allocate a writable empty string since parse_monitor_columns()
+         * is going to strtok() it and that's risky with literal "". */
         char empty[] = "";
         json_array_add(
             monitor_request_array,
-            parse_monitor_columns(empty, server, database, table, &columns));
+            parse_monitor_columns(empty, server, database,
+                                  table, &mt->columns));
+    }
+
+    json_object_put(monitor_requests, table->name, monitor_request_array);
+}
+
+static void
+do_monitor(struct jsonrpc *rpc, const char *database,
+           int argc, char *argv[])
+{
+    const char *server = jsonrpc_get_name(rpc);
+    const char *table_name = argv[0];
+    struct unixctl_server *unixctl;
+    struct ovsdb_schema *schema;
+    struct jsonrpc_msg *request;
+    struct json *monitor, *monitor_requests, *request_id;
+    bool exiting = false;
+    bool blocked = false;
+
+    struct monitored_table *mts;
+    size_t n_mts, allocated_mts;
+
+    daemon_save_fd(STDOUT_FILENO);
+    daemonize_start();
+    if (get_detach()) {
+        int error;
+
+        error = unixctl_server_create(NULL, &unixctl);
+        if (error) {
+            ovs_fatal(error, "failed to create unixctl server");
+        }
+
+        unixctl_command_register("exit", "", 0, 0,
+                                 ovsdb_client_exit, &exiting);
+        unixctl_command_register("ovsdb-client/block", "", 0, 0,
+                                 ovsdb_client_block, &blocked);
+        unixctl_command_register("ovsdb-client/unblock", "", 0, 0,
+                                 ovsdb_client_unblock, &blocked);
+    } else {
+        unixctl = NULL;
     }
 
+    schema = fetch_schema(rpc, database);
+
     monitor_requests = json_object_create();
-    json_object_put(monitor_requests, table_name, monitor_request_array);
+
+    mts = NULL;
+    n_mts = allocated_mts = 0;
+    if (strcmp(table_name, "ALL")) {
+        struct ovsdb_table_schema *table;
+
+        table = shash_find_data(&schema->tables, table_name);
+        if (!table) {
+            ovs_fatal(0, "%s: %s does not have a table named \"%s\"",
+                      server, database, table_name);
+        }
+
+        add_monitored_table(argc, argv, server, database, table,
+                            monitor_requests, &mts, &n_mts, &allocated_mts);
+    } else {
+        size_t n = shash_count(&schema->tables);
+        const struct shash_node **nodes = shash_sort(&schema->tables);
+        size_t i;
+
+        for (i = 0; i < n; i++) {
+            struct ovsdb_table_schema *table = nodes[i]->data;
+
+            add_monitored_table(argc, argv, server, database, table,
+                                monitor_requests,
+                                &mts, &n_mts, &allocated_mts);
+        }
+        free(nodes);
+    }
 
     monitor = json_array_create_3(json_string_create(database),
                                   json_null_create(), monitor_requests);
     request = jsonrpc_create_request("monitor", monitor, NULL);
     request_id = json_clone(request->id);
     jsonrpc_send(rpc, request);
-    for (;;) {
-        struct jsonrpc_msg *msg;
-        int error;
-
-        error = jsonrpc_recv_block(rpc, &msg);
-        if (error) {
-            ovsdb_schema_destroy(schema);
-            ovs_fatal(error, "%s: receive failed", server);
-        }
 
-        if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
-            jsonrpc_send(rpc, jsonrpc_create_reply(json_clone(msg->params),
-                                                   msg->id));
-        } else if (msg->type == JSONRPC_REPLY
-                   && json_equal(msg->id, request_id)) {
-            monitor_print(msg->result, table, &columns, true);
-            fflush(stdout);
-            if (get_detach()) {
-                /* daemonize() closes the standard file descriptors.  We output
-                 * to stdout, so we need to save and restore STDOUT_FILENO. */
-                int fd = dup(STDOUT_FILENO);
-                daemonize();
-                dup2(fd, STDOUT_FILENO);
-                close(fd);
+    for (;;) {
+        unixctl_server_run(unixctl);
+        while (!blocked) {
+            struct jsonrpc_msg *msg;
+            int error;
+
+            error = jsonrpc_recv(rpc, &msg);
+            if (error == EAGAIN) {
+                break;
+            } else if (error) {
+                ovs_fatal(error, "%s: receive failed", server);
             }
-        } else if (msg->type == JSONRPC_NOTIFY
-                   && !strcmp(msg->method, "update")) {
-            struct json *params = msg->params;
-            if (params->type == JSON_ARRAY
-                && params->u.array.n == 2
-                && params->u.array.elems[0]->type == JSON_NULL) {
-                monitor_print(params->u.array.elems[1],
-                              table, &columns, false);
+
+            if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
+                jsonrpc_send(rpc, jsonrpc_create_reply(json_clone(msg->params),
+                                                       msg->id));
+            } else if (msg->type == JSONRPC_REPLY
+                       && json_equal(msg->id, request_id)) {
+                monitor_print(msg->result, mts, n_mts, true);
                 fflush(stdout);
+                daemonize_complete();
+            } else if (msg->type == JSONRPC_NOTIFY
+                       && !strcmp(msg->method, "update")) {
+                struct json *params = msg->params;
+                if (params->type == JSON_ARRAY
+                    && params->u.array.n == 2
+                    && params->u.array.elems[0]->type == JSON_NULL) {
+                    monitor_print(params->u.array.elems[1], mts, n_mts, false);
+                    fflush(stdout);
+                }
             }
+            jsonrpc_msg_destroy(msg);
+        }
+
+        if (exiting) {
+            break;
+        }
+
+        jsonrpc_run(rpc);
+        jsonrpc_wait(rpc);
+        if (!blocked) {
+            jsonrpc_recv_wait(rpc);
         }
-        jsonrpc_msg_destroy(msg);
+        unixctl_server_wait(unixctl);
+        poll_block();
     }
 }
 
@@ -749,7 +991,7 @@ dump_table(const struct ovsdb_table_schema *ts, struct json_array *rows)
         struct shash *row;
 
         if (rows->elems[y]->type != JSON_OBJECT) {
-            ovs_fatal(0,  "row %zu in table %s response is not a JSON object: "
+            ovs_fatal(0,  "row %"PRIuSIZE" in table %s response is not a JSON object: "
                       "%s", y, ts->name, json_to_string(rows->elems[y], 0));
         }
         row = json_object(rows->elems[y]);
@@ -758,7 +1000,7 @@ dump_table(const struct ovsdb_table_schema *ts, struct json_array *rows)
         for (x = 0; x < n_columns; x++) {
             const struct json *json = shash_find_data(row, columns[x]->name);
             if (!json) {
-                ovs_fatal(0, "row %zu in table %s response lacks %s column",
+                ovs_fatal(0, "row %"PRIuSIZE" in table %s response lacks %s column",
                           y, ts->name, columns[x]->name);
             }
 
@@ -788,32 +1030,31 @@ dump_table(const struct ovsdb_table_schema *ts, struct json_array *rows)
             struct cell *cell = table_add_cell(&t);
             cell->json = ovsdb_datum_to_json(&data[y][x], &columns[x]->type);
             cell->type = &columns[x]->type;
+            ovsdb_datum_destroy(&data[y][x], &columns[x]->type);
         }
+        free(data[y]);
     }
     table_print(&t, &table_style);
     table_destroy(&t);
+
+    free(data);
+    free(columns);
 }
 
 static void
-do_dump(int argc OVS_UNUSED, char *argv[])
+do_dump(struct jsonrpc *rpc, const char *database,
+        int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
 {
-    const char *server = argv[1];
-    const char *database = argv[2];
-
     struct jsonrpc_msg *request, *reply;
     struct ovsdb_schema *schema;
     struct json *transaction;
-    struct jsonrpc *rpc;
-    int error;
 
     const struct shash_node **tables;
     size_t n_tables;
 
     size_t i;
 
-    rpc = open_jsonrpc(server);
-
-    schema = fetch_schema_from_rpc(rpc, database);
+    schema = fetch_schema(rpc, database);
     tables = shash_sort(&schema->tables);
     n_tables = shash_count(&schema->tables);
 
@@ -843,15 +1084,12 @@ do_dump(int argc OVS_UNUSED, char *argv[])
 
     /* Send request, get reply. */
     request = jsonrpc_create_request("transact", transaction, NULL);
-    error = jsonrpc_transact_block(rpc, request, &reply);
-    if (error) {
-        ovs_fatal(error, "transaction failed");
-    }
+    check_txn(jsonrpc_transact_block(rpc, request, &reply), &reply);
 
     /* Print database contents. */
     if (reply->result->type != JSON_ARRAY
         || reply->result->u.array.n != n_tables) {
-        ovs_fatal(0, "reply is not array of %zu elements: %s",
+        ovs_fatal(0, "reply is not array of %"PRIuSIZE" elements: %s",
                   n_tables, json_to_string(reply->result, 0));
     }
     for (i = 0; i < n_tables; i++) {
@@ -869,23 +1107,42 @@ do_dump(int argc OVS_UNUSED, char *argv[])
 
         dump_table(ts, &rows->u.array);
     }
+
+    jsonrpc_msg_destroy(reply);
+    free(tables);
+    ovsdb_schema_destroy(schema);
 }
 
 static void
-do_help(int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
+do_help(struct jsonrpc *rpc OVS_UNUSED, const char *database OVS_UNUSED,
+        int argc OVS_UNUSED, char *argv[] OVS_UNUSED)
 {
     usage();
 }
 
-static const struct command all_commands[] = {
-    { "list-dbs", 1, 1, do_list_dbs },
-    { "get-schema", 2, 2, do_get_schema },
-    { "get-schema-version", 2, 2, do_get_schema_version },
-    { "list-tables", 2, 2, do_list_tables },
-    { "list-columns", 2, 3, do_list_columns },
-    { "transact", 2, 2, do_transact },
-    { "monitor", 3, INT_MAX, do_monitor },
-    { "dump", 2, 2, do_dump },
-    { "help", 0, INT_MAX, do_help },
-    { NULL, 0, 0, NULL },
+/* All command handlers (except for "help") are expected to take an optional
+ * server socket name (e.g. "unix:...") as their first argument.  The socket
+ * name argument must be included in max_args (but left out of min_args).  The
+ * command name and socket name are not included in the arguments passed to the
+ * handler: the argv[0] passed to the handler is the first argument after the
+ * optional server socket name.  The connection to the server is available as
+ * global variable 'rpc'. */
+static const struct ovsdb_client_command all_commands[] = {
+    { "list-dbs",           NEED_RPC,      0, 0,       do_list_dbs },
+    { "get-schema",         NEED_DATABASE, 0, 0,       do_get_schema },
+    { "get-schema-version", NEED_DATABASE, 0, 0,       do_get_schema_version },
+    { "list-tables",        NEED_DATABASE, 0, 0,       do_list_tables },
+    { "list-columns",       NEED_DATABASE, 0, 1,       do_list_columns },
+    { "transact",           NEED_RPC,      1, 1,       do_transact },
+    { "monitor",            NEED_DATABASE, 1, INT_MAX, do_monitor },
+    { "dump",               NEED_DATABASE, 0, 0,       do_dump },
+
+    { "help",               NEED_NONE,     0, INT_MAX, do_help },
+
+    { NULL,                 0,             0, 0,       NULL },
 };
+
+static const struct ovsdb_client_command *get_all_commands(void)
+{
+    return all_commands;
+}