From: Ethan Jackson <ethan@nicira.com>
Date: Wed, 15 Feb 2012 04:53:59 +0000 (-0800)
Subject: unixctl: New JSON RPC back-end.
X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=bde9f75de;p=openvswitch

unixctl: New JSON RPC back-end.

The unixctl library had used the vde2 management protocol since the
early days of Open vSwitch.  As Open vSwitch has matured, several
Python daemons have been added to the code base which would benefit
from a unixctl implementations.  Instead of implementing the old
unixctl protocol in Python, this patch changes unixctl to use JSON
RPC for which we already have an implementation in both Python and
C.  Future patches will need to implement a unixctl library in
Python on top of JSON RPC.

Signed-off-by: Ethan Jackson <ethan@nicira.com>
---

diff --git a/lib/bond.c b/lib/bond.c
index 1ad2367d..3b86ee48 100644
--- a/lib/bond.c
+++ b/lib/bond.c
@@ -941,7 +941,7 @@ bond_unixctl_list(struct unixctl_conn *conn,
         }
         ds_put_char(&ds, '\n');
     }
-    unixctl_command_reply(conn, 200, ds_cstr(&ds));
+    unixctl_command_reply(conn, ds_cstr(&ds));
     ds_destroy(&ds);
 }
 
@@ -1042,7 +1042,7 @@ bond_unixctl_show(struct unixctl_conn *conn,
         const struct bond *bond = bond_find(argv[1]);
 
         if (!bond) {
-            unixctl_command_reply(conn, 501, "no such bond");
+            unixctl_command_reply_error(conn, "no such bond");
             return;
         }
         bond_print_details(&ds, bond);
@@ -1054,7 +1054,7 @@ bond_unixctl_show(struct unixctl_conn *conn,
         }
     }
 
-    unixctl_command_reply(conn, 200, ds_cstr(&ds));
+    unixctl_command_reply(conn, ds_cstr(&ds));
     ds_destroy(&ds);
 }
 
@@ -1073,30 +1073,30 @@ bond_unixctl_migrate(struct unixctl_conn *conn,
 
     bond = bond_find(bond_s);
     if (!bond) {
-        unixctl_command_reply(conn, 501, "no such bond");
+        unixctl_command_reply_error(conn, "no such bond");
         return;
     }
 
     if (bond->balance != BM_SLB) {
-        unixctl_command_reply(conn, 501, "not an SLB bond");
+        unixctl_command_reply_error(conn, "not an SLB bond");
         return;
     }
 
     if (strspn(hash_s, "0123456789") == strlen(hash_s)) {
         hash = atoi(hash_s) & BOND_MASK;
     } else {
-        unixctl_command_reply(conn, 501, "bad hash");
+        unixctl_command_reply_error(conn, "bad hash");
         return;
     }
 
     slave = bond_lookup_slave(bond, slave_s);
     if (!slave) {
-        unixctl_command_reply(conn, 501, "no such slave");
+        unixctl_command_reply_error(conn, "no such slave");
         return;
     }
 
     if (!slave->enabled) {
-        unixctl_command_reply(conn, 501, "cannot migrate to disabled slave");
+        unixctl_command_reply_error(conn, "cannot migrate to disabled slave");
         return;
     }
 
@@ -1104,7 +1104,7 @@ bond_unixctl_migrate(struct unixctl_conn *conn,
     tag_set_add(&bond->unixctl_tags, entry->tag);
     entry->slave = slave;
     entry->tag = tag_create_random();
-    unixctl_command_reply(conn, 200, "migrated");
+    unixctl_command_reply(conn, "migrated");
 }
 
 static void
@@ -1119,18 +1119,18 @@ bond_unixctl_set_active_slave(struct unixctl_conn *conn,
 
     bond = bond_find(bond_s);
     if (!bond) {
-        unixctl_command_reply(conn, 501, "no such bond");
+        unixctl_command_reply_error(conn, "no such bond");
         return;
     }
 
     slave = bond_lookup_slave(bond, slave_s);
     if (!slave) {
-        unixctl_command_reply(conn, 501, "no such slave");
+        unixctl_command_reply_error(conn, "no such slave");
         return;
     }
 
     if (!slave->enabled) {
-        unixctl_command_reply(conn, 501, "cannot make disabled slave active");
+        unixctl_command_reply_error(conn, "cannot make disabled slave active");
         return;
     }
 
@@ -1141,9 +1141,9 @@ bond_unixctl_set_active_slave(struct unixctl_conn *conn,
         VLOG_INFO("bond %s: active interface is now %s",
                   bond->name, slave->name);
         bond->send_learning_packets = true;
-        unixctl_command_reply(conn, 200, "done");
+        unixctl_command_reply(conn, "done");
     } else {
-        unixctl_command_reply(conn, 200, "no change");
+        unixctl_command_reply(conn, "no change");
     }
 }
 
@@ -1157,18 +1157,18 @@ enable_slave(struct unixctl_conn *conn, const char *argv[], bool enable)
 
     bond = bond_find(bond_s);
     if (!bond) {
-        unixctl_command_reply(conn, 501, "no such bond");
+        unixctl_command_reply_error(conn, "no such bond");
         return;
     }
 
     slave = bond_lookup_slave(bond, slave_s);
     if (!slave) {
-        unixctl_command_reply(conn, 501, "no such slave");
+        unixctl_command_reply_error(conn, "no such slave");
         return;
     }
 
     bond_enable_slave(slave, enable, &bond->unixctl_tags);
-    unixctl_command_reply(conn, 200, enable ? "enabled" : "disabled");
+    unixctl_command_reply(conn, enable ? "enabled" : "disabled");
 }
 
 static void
@@ -1202,7 +1202,7 @@ bond_unixctl_hash(struct unixctl_conn *conn, int argc, const char *argv[],
 
     if (vlan_s) {
         if (sscanf(vlan_s, "%u", &vlan) != 1) {
-            unixctl_command_reply(conn, 501, "invalid vlan");
+            unixctl_command_reply_error(conn, "invalid vlan");
             return;
         }
     } else {
@@ -1211,7 +1211,7 @@ bond_unixctl_hash(struct unixctl_conn *conn, int argc, const char *argv[],
 
     if (basis_s) {
         if (sscanf(basis_s, "%"PRIu32, &basis) != 1) {
-            unixctl_command_reply(conn, 501, "invalid basis");
+            unixctl_command_reply_error(conn, "invalid basis");
             return;
         }
     } else {
@@ -1223,10 +1223,10 @@ bond_unixctl_hash(struct unixctl_conn *conn, int argc, const char *argv[],
         hash = bond_hash_src(mac, vlan, basis) & BOND_MASK;
 
         hash_cstr = xasprintf("%u", hash);
-        unixctl_command_reply(conn, 200, hash_cstr);
+        unixctl_command_reply(conn, hash_cstr);
         free(hash_cstr);
     } else {
-        unixctl_command_reply(conn, 501, "invalid mac");
+        unixctl_command_reply_error(conn, "invalid mac");
     }
 }
 
diff --git a/lib/cfm.c b/lib/cfm.c
index d373f422..8d90829a 100644
--- a/lib/cfm.c
+++ b/lib/cfm.c
@@ -677,7 +677,7 @@ cfm_unixctl_show(struct unixctl_conn *conn, int argc, const char *argv[],
     if (argc > 1) {
         cfm = cfm_find(argv[1]);
         if (!cfm) {
-            unixctl_command_reply(conn, 501, "no such CFM object");
+            unixctl_command_reply_error(conn, "no such CFM object");
             return;
         }
         cfm_print_details(&ds, cfm);
@@ -687,7 +687,7 @@ cfm_unixctl_show(struct unixctl_conn *conn, int argc, const char *argv[],
         }
     }
 
-    unixctl_command_reply(conn, 200, ds_cstr(&ds));
+    unixctl_command_reply(conn, ds_cstr(&ds));
     ds_destroy(&ds);
 }
 
@@ -706,14 +706,14 @@ cfm_unixctl_set_fault(struct unixctl_conn *conn, int argc, const char *argv[],
     } else if (!strcasecmp("normal", fault_str)) {
         fault_override = -1;
     } else {
-        unixctl_command_reply(conn, 501, "unknown fault string");
+        unixctl_command_reply_error(conn, "unknown fault string");
         return;
     }
 
     if (argc > 2) {
         cfm = cfm_find(argv[1]);
         if (!cfm) {
-            unixctl_command_reply(conn, 501, "no such CFM object");
+            unixctl_command_reply_error(conn, "no such CFM object");
             return;
         }
         cfm->fault_override = fault_override;
@@ -723,5 +723,5 @@ cfm_unixctl_set_fault(struct unixctl_conn *conn, int argc, const char *argv[],
         }
     }
 
-    unixctl_command_reply(conn, 200, "OK");
+    unixctl_command_reply(conn, "OK");
 }
diff --git a/lib/coverage.c b/lib/coverage.c
index 08b49610..0deb5265 100644
--- a/lib/coverage.c
+++ b/lib/coverage.c
@@ -53,7 +53,7 @@ coverage_unixctl_log(struct unixctl_conn *conn, int argc OVS_UNUSED,
                      const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
 {
     coverage_log(VLL_WARN, false);
-    unixctl_command_reply(conn, 200, NULL);
+    unixctl_command_reply(conn, NULL);
 }
 
 void
diff --git a/lib/lacp.c b/lib/lacp.c
index c51fac69..f7d5a82f 100644
--- a/lib/lacp.c
+++ b/lib/lacp.c
@@ -878,7 +878,7 @@ lacp_unixctl_show(struct unixctl_conn *conn, int argc, const char *argv[],
     if (argc > 1) {
         lacp = lacp_find(argv[1]);
         if (!lacp) {
-            unixctl_command_reply(conn, 501, "no such lacp object");
+            unixctl_command_reply_error(conn, "no such lacp object");
             return;
         }
         lacp_print_details(&ds, lacp);
@@ -888,6 +888,6 @@ lacp_unixctl_show(struct unixctl_conn *conn, int argc, const char *argv[],
         }
     }
 
-    unixctl_command_reply(conn, 200, ds_cstr(&ds));
+    unixctl_command_reply(conn, ds_cstr(&ds));
     ds_destroy(&ds);
 }
diff --git a/lib/netdev-dummy.c b/lib/netdev-dummy.c
index 57962d10..5c6563cc 100644
--- a/lib/netdev-dummy.c
+++ b/lib/netdev-dummy.c
@@ -410,7 +410,7 @@ netdev_dummy_receive(struct unixctl_conn *conn,
 
     dummy_dev = shash_find_data(&dummy_netdev_devs, argv[1]);
     if (!dummy_dev) {
-        unixctl_command_reply(conn, 501, "no such dummy netdev");
+        unixctl_command_reply_error(conn, "no such dummy netdev");
         return;
     }
 
@@ -421,7 +421,7 @@ netdev_dummy_receive(struct unixctl_conn *conn,
 
         packet = eth_from_packet_or_flow(argv[i]);
         if (!packet) {
-            unixctl_command_reply(conn, 501, "bad packet syntax");
+            unixctl_command_reply_error(conn, "bad packet syntax");
             return;
         }
 
@@ -437,9 +437,9 @@ netdev_dummy_receive(struct unixctl_conn *conn,
     }
 
     if (!n_listeners) {
-        unixctl_command_reply(conn, 202, "packets queued but nobody listened");
+        unixctl_command_reply(conn, "packets queued but nobody listened");
     } else {
-        unixctl_command_reply(conn, 200, "success");
+        unixctl_command_reply(conn, "success");
     }
 }
 
diff --git a/lib/stp.c b/lib/stp.c
index 06ac0019..e9230531 100644
--- a/lib/stp.c
+++ b/lib/stp.c
@@ -1371,7 +1371,7 @@ stp_unixctl_tcn(struct unixctl_conn *conn, int argc,
         struct stp *stp = stp_find(argv[1]);
 
         if (!stp) {
-            unixctl_command_reply(conn, 501, "no such stp object");
+            unixctl_command_reply_error(conn, "no such stp object");
             return;
         }
         stp_topology_change_detection(stp);
@@ -1383,5 +1383,5 @@ stp_unixctl_tcn(struct unixctl_conn *conn, int argc,
         }
     }
 
-    unixctl_command_reply(conn, 200, "OK");
+    unixctl_command_reply(conn, "OK");
 }
diff --git a/lib/stress.c b/lib/stress.c
index 0fdc79aa..dfd00990 100644
--- a/lib/stress.c
+++ b/lib/stress.c
@@ -154,9 +154,9 @@ stress_unixctl_list(struct unixctl_conn *conn, int argc OVS_UNUSED,
         }
     }
     if (found) {
-        unixctl_command_reply(conn, 200, ds_cstr(&results));
+        unixctl_command_reply(conn, ds_cstr(&results));
     } else {
-        unixctl_command_reply(conn, 404, "");
+        unixctl_command_reply_error(conn, NULL);
     }
     ds_destroy(&results);
 }
@@ -166,7 +166,7 @@ stress_unixctl_enable(struct unixctl_conn *conn, int argc OVS_UNUSED,
                       const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
 {
     stress_enable(true);
-    unixctl_command_reply(conn, 200, "");
+    unixctl_command_reply(conn, NULL);
 }
 
 static void
@@ -174,14 +174,13 @@ stress_unixctl_disable(struct unixctl_conn *conn, int argc OVS_UNUSED,
                        const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
 {
     stress_enable(false);
-    unixctl_command_reply(conn, 200, "");
+    unixctl_command_reply(conn, NULL);
 }
 
 static void
 stress_unixctl_set(struct unixctl_conn *conn, int argc OVS_UNUSED,
                    const char *argv[], void *aux OVS_UNUSED)
 {
-    int code = 404;
     const char *option_name = argv[1];
     const char *option_val = argv[2];
     int i;
@@ -193,11 +192,12 @@ stress_unixctl_set(struct unixctl_conn *conn, int argc OVS_UNUSED,
             bool random = !strcmp(argv[3], "random");
 
             stress_set(option, period, random);
-            code = 200;
-            break;
+            unixctl_command_reply(conn, NULL);
+            return;
         }
     }
-    unixctl_command_reply(conn, code, "");
+
+    unixctl_command_reply_error(conn, NULL);
 }
 
 /* Exposes ovs-appctl access to the stress options.
diff --git a/lib/timeval.c b/lib/timeval.c
index 8097ce80..35819edd 100644
--- a/lib/timeval.c
+++ b/lib/timeval.c
@@ -570,14 +570,14 @@ timeval_warp_cb(struct unixctl_conn *conn,
 
     msecs = atoi(argv[1]);
     if (msecs <= 0) {
-        unixctl_command_reply(conn, 501, "invalid MSECS");
+        unixctl_command_reply_error(conn, "invalid MSECS");
         return;
     }
 
     ts.tv_sec = msecs / 1000;
     ts.tv_nsec = (msecs % 1000) * 1000 * 1000;
     timespec_add(&warp_offset, &warp_offset, &ts);
-    unixctl_command_reply(conn, 200, "warped");
+    unixctl_command_reply(conn, "warped");
 }
 
 void
diff --git a/lib/unixctl.c b/lib/unixctl.c
index caaf252d..054ce49c 100644
--- a/lib/unixctl.c
+++ b/lib/unixctl.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008, 2009, 2010, 2011 Nicira Networks.
+ * Copyright (c) 2008, 2009, 2010, 2011, 2012 Nicira Networks.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,31 +17,20 @@
 #include <config.h>
 #include "unixctl.h"
 #include <assert.h>
-#include <ctype.h>
 #include <errno.h>
-#include <poll.h>
-#include <stdlib.h>
-#include <string.h>
-#include <sys/socket.h>
-#include <sys/stat.h>
 #include <unistd.h>
 #include "coverage.h"
 #include "dirs.h"
 #include "dynamic-string.h"
-#include "fatal-signal.h"
+#include "json.h"
+#include "jsonrpc.h"
 #include "list.h"
-#include "ofpbuf.h"
 #include "poll-loop.h"
 #include "shash.h"
-#include "socket-util.h"
+#include "stream.h"
 #include "svec.h"
-#include "util.h"
 #include "vlog.h"
 
-#ifndef SCM_CREDENTIALS
-#include <time.h>
-#endif
-
 VLOG_DEFINE_THIS_MODULE(unixctl);
 
 COVERAGE_DEFINE(unixctl_received);
@@ -56,28 +45,19 @@ struct unixctl_command {
 
 struct unixctl_conn {
     struct list node;
-    int fd;
+    struct jsonrpc *rpc;
 
-    enum { S_RECV, S_PROCESS, S_SEND } state;
-    struct ofpbuf in;
-    struct ds out;
-    size_t out_pos;
+    /* Only one request can be in progress at a time.  While the request is
+     * being processed, 'request_id' is populated, otherwise it is null. */
+    struct json *request_id;   /* ID of the currently active request. */
 };
 
 /* Server for control connection. */
 struct unixctl_server {
-    char *path;
-    int fd;
+    struct pstream *listener;
     struct list conns;
 };
 
-/* Client for control connection. */
-struct unixctl_client {
-    char *connect_path;
-    char *bind_path;
-    FILE *stream;
-};
-
 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
 
 static struct shash commands = SHASH_INITIALIZER(&commands);
@@ -95,12 +75,12 @@ unixctl_help(struct unixctl_conn *conn, int argc OVS_UNUSED,
     for (i = 0; i < shash_count(&commands); i++) {
         const struct shash_node *node = nodes[i];
         const struct unixctl_command *command = node->data;
-        
+
         ds_put_format(&ds, "  %-23s %s\n", node->name, command->usage);
     }
     free(nodes);
 
-    unixctl_command_reply(conn, 214, ds_cstr(&ds));
+    unixctl_command_reply(conn, ds_cstr(&ds));
     ds_destroy(&ds);
 }
 
@@ -108,7 +88,7 @@ static void
 unixctl_version(struct unixctl_conn *conn, int argc OVS_UNUSED,
                 const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
 {
-    unixctl_command_reply(conn, 200, get_program_version());
+    unixctl_command_reply(conn, get_program_version());
 }
 
 /* Registers a unixctl command with the given 'name'.  'usage' describes the
@@ -117,11 +97,11 @@ unixctl_version(struct unixctl_conn *conn, int argc OVS_UNUSED,
  *
  * 'cb' is called when the command is received.  It is passed the actual set of
  * arguments, as a text string, plus a copy of 'aux'.  Normally 'cb' should
- * call unixctl_command_reply() before it returns, but if the command cannot be
- * handled immediately then it can defer the reply until later.  A given
- * connection can only process a single request at a time, so
- * unixctl_command_reply() must be called eventually to avoid blocking that
- * connection. */
+ * reply by calling unixctl_command_reply() or unixctl_command_reply_error()
+ * before it returns, but if the command cannot be handled immediately then it
+ * can defer the reply until later.  A given connection can only process a
+ * single request at a time, so a reply must be made eventually to avoid
+ * blocking that connection. */
 void
 unixctl_command_register(const char *name, const char *usage,
                          int min_args, int max_args,
@@ -145,57 +125,57 @@ unixctl_command_register(const char *name, const char *usage,
     shash_add(&commands, name, command);
 }
 
-static const char *
-translate_reply_code(int code)
+static void
+unixctl_command_reply__(struct unixctl_conn *conn,
+                        bool success, const char *body)
 {
-    switch (code) {
-    case 200: return "OK";
-    case 201: return "Created";
-    case 202: return "Accepted";
-    case 204: return "No Content";
-    case 211: return "System Status";
-    case 214: return "Help";
-    case 400: return "Bad Request";
-    case 401: return "Unauthorized";
-    case 403: return "Forbidden";
-    case 404: return "Not Found";
-    case 500: return "Internal Server Error";
-    case 501: return "Invalid Argument";
-    case 503: return "Service Unavailable";
-    default: return "Unknown";
+    struct json *body_json;
+    struct jsonrpc_msg *reply;
+
+    COVERAGE_INC(unixctl_replied);
+    assert(conn->request_id);
+
+    if (!body) {
+        body = "";
     }
+
+    if (body[0] && body[strlen(body) - 1] != '\n') {
+        body_json = json_string_create_nocopy(xasprintf("%s\n", body));
+    } else {
+        body_json = json_string_create(body);
+    }
+
+    if (success) {
+        reply = jsonrpc_create_reply(body_json, conn->request_id);
+    } else {
+        reply = jsonrpc_create_error(body_json, conn->request_id);
+    }
+
+    /* If jsonrpc_send() returns an error, the run loop will take care of the
+     * problem eventually. */
+    jsonrpc_send(conn->rpc, reply);
+    json_destroy(conn->request_id);
+    conn->request_id = NULL;
 }
 
+/* Replies to the active unixctl connection 'conn'.  'result' is sent to the
+ * client indicating the command was processed successfully.  Only one call to
+ * unixctl_command_reply() or unixctl_command_reply_error() may be made per
+ * request. */
 void
-unixctl_command_reply(struct unixctl_conn *conn,
-                      int code, const char *body)
+unixctl_command_reply(struct unixctl_conn *conn, const char *result)
 {
-    struct ds *out = &conn->out;
+    unixctl_command_reply__(conn, true, result);
+}
 
-    COVERAGE_INC(unixctl_replied);
-    assert(conn->state == S_PROCESS);
-    conn->state = S_SEND;
-    conn->out_pos = 0;
-
-    ds_clear(out);
-    ds_put_format(out, "%03d %s\n", code, translate_reply_code(code));
-    if (body) {
-        const char *p;
-        for (p = body; *p != '\0'; ) {
-            size_t n = strcspn(p, "\n");
-
-            if (*p == '.') {
-                ds_put_char(out, '.');
-            }
-            ds_put_buffer(out, p, n);
-            ds_put_char(out, '\n');
-            p += n;
-            if (*p == '\n') {
-                p++;
-            }
-        }
-    }
-    ds_put_cstr(out, ".\n");
+/* Replies to the active unixctl connection 'conn'. 'error' is sent to the
+ * client indicating an error occured processing the command.  Only one call to
+ * unixctl_command_reply() or unixctl_command_reply_error() may be made per
+ * request. */
+void
+unixctl_command_reply_error(struct unixctl_conn *conn, const char *error)
+{
+    unixctl_command_reply__(conn, false, error);
 }
 
 /* Creates a unixctl server listening on 'path', which may be:
@@ -223,229 +203,135 @@ int
 unixctl_server_create(const char *path, struct unixctl_server **serverp)
 {
     struct unixctl_server *server;
+    struct pstream *listener;
+    char *punix_path;
     int error;
 
+    *serverp = NULL;
     if (path && !strcmp(path, "none")) {
-        *serverp = NULL;
         return 0;
     }
 
-    unixctl_command_register("help", "", 0, 0, unixctl_help, NULL);
-    unixctl_command_register("version", "", 0, 0, unixctl_version, NULL);
-
-    server = xmalloc(sizeof *server);
-    list_init(&server->conns);
-
     if (path) {
-        server->path = abs_file_name(ovs_rundir(), path);
+        char *abs_path = abs_file_name(ovs_rundir(), path);
+        punix_path = xasprintf("punix:%s", abs_path);
+        free(abs_path);
     } else {
-        server->path = xasprintf("%s/%s.%ld.ctl", ovs_rundir(),
-                                 program_name, (long int) getpid());
+        punix_path = xasprintf("punix:%s/%s.%ld.ctl", ovs_rundir(),
+                               program_name, (long int) getpid());
     }
 
-    server->fd = make_unix_socket(SOCK_STREAM, true, false, server->path,
-                                  NULL);
-    if (server->fd < 0) {
-        error = -server->fd;
-        ovs_error(error, "could not initialize control socket %s",
-                  server->path);
-        goto error;
-    }
+    error = pstream_open(punix_path, &listener);
+    free(punix_path);
+    punix_path = NULL;
 
-    if (chmod(server->path, S_IRUSR | S_IWUSR) < 0) {
-        error = errno;
-        ovs_error(error, "failed to chmod control socket %s", server->path);
-        goto error;
+    if (error) {
+        ovs_error(error, "could not initialize control socket %s", path);
+        return error;
     }
 
-    if (listen(server->fd, 10) < 0) {
-        error = errno;
-        ovs_error(error, "Failed to listen on control socket %s",
-                  server->path);
-        goto error;
-    }
+    unixctl_command_register("help", "", 0, 0, unixctl_help, NULL);
+    unixctl_command_register("version", "", 0, 0, unixctl_version, NULL);
 
+    server = xmalloc(sizeof *server);
+    server->listener = listener;
+    list_init(&server->conns);
     *serverp = server;
     return 0;
-
-error:
-    if (server->fd >= 0) {
-        close(server->fd);
-    }
-    free(server->path);
-    free(server);
-    *serverp = NULL;
-    return error;
 }
 
 static void
-new_connection(struct unixctl_server *server, int fd)
-{
-    struct unixctl_conn *conn;
-
-    set_nonblocking(fd);
-
-    conn = xmalloc(sizeof *conn);
-    list_push_back(&server->conns, &conn->node);
-    conn->fd = fd;
-    conn->state = S_RECV;
-    ofpbuf_init(&conn->in, 128);
-    ds_init(&conn->out);
-    conn->out_pos = 0;
-}
-
-static int
-run_connection_output(struct unixctl_conn *conn)
+process_command(struct unixctl_conn *conn, struct jsonrpc_msg *request)
 {
-    while (conn->out_pos < conn->out.length) {
-        size_t bytes_written;
-        int error;
-
-        error = write_fully(conn->fd, conn->out.string + conn->out_pos,
-                            conn->out.length - conn->out_pos, &bytes_written);
-        conn->out_pos += bytes_written;
-        if (error) {
-            return error;
-        }
-    }
-    conn->state = S_RECV;
-    return 0;
-}
+    char *error = NULL;
 
-static void
-process_command(struct unixctl_conn *conn, char *s)
-{
     struct unixctl_command *command;
-    struct svec argv;
+    struct json_array *params;
 
     COVERAGE_INC(unixctl_received);
-    conn->state = S_PROCESS;
-
-    svec_init(&argv);
-    svec_parse_words(&argv, s);
-    svec_terminate(&argv);
-
-    if (argv.n == 0) {
-        unixctl_command_reply(conn, 400, "missing command name in request");
+    conn->request_id = json_clone(request->id);
+
+    params = json_array(request->params);
+    command = shash_find_data(&commands, request->method);
+    if (!command) {
+        error = xasprintf("\"%s\" is not a valid command", request->method);
+    } else if (params->n < command->min_args) {
+        error = xasprintf("\"%s\" command requires at least %d arguments",
+                          request->method, command->min_args);
+    } else if (params->n > command->max_args) {
+        error = xasprintf("\"%s\" command takes at most %d arguments",
+                          request->method, command->max_args);
     } else {
-        const char *name = argv.names[0];
-        char *error;
-
-        command = shash_find_data(&commands, name);
-        if (!command) {
-            error = xasprintf("\"%s\" is not a valid command", name);
-        } else if (argv.n - 1 < command->min_args) {
-            error = xasprintf("\"%s\" command requires at least %d arguments",
-                              name, command->min_args);
-        } else if (argv.n - 1 > command->max_args) {
-            error = xasprintf("\"%s\" command takes at most %d arguments",
-                              name, command->max_args);
-        } else {
-            error = NULL;
+        struct svec argv = SVEC_EMPTY_INITIALIZER;
+        int  i;
+
+        svec_add(&argv, request->method);
+        for (i = 0; i < params->n; i++) {
+            if (params->elems[i]->type != JSON_STRING) {
+                error = xasprintf("\"%s\" command has non-string argument",
+                                  request->method);
+                break;
+            }
+            svec_add(&argv, json_string(params->elems[i]));
+        }
+        svec_terminate(&argv);
+
+        if (!error) {
             command->cb(conn, argv.n, (const char **) argv.names,
                         command->aux);
         }
 
-        if (error) {
-            unixctl_command_reply(conn, 400, error);
-            free(error);
-        }
+        svec_destroy(&argv);
     }
 
-    svec_destroy(&argv);
+    if (error) {
+        unixctl_command_reply_error(conn, error);
+        free(error);
+    }
 }
 
 static int
-run_connection_input(struct unixctl_conn *conn)
+run_connection(struct unixctl_conn *conn)
 {
-    for (;;) {
-        size_t bytes_read;
-        char *newline;
-        int error;
-
-        newline = memchr(conn->in.data, '\n', conn->in.size);
-        if (newline) {
-            char *command = conn->in.data;
-            size_t n = newline - command + 1;
+    int error, i;
 
-            if (n > 0 && newline[-1] == '\r') {
-                newline--;
-            }
-            *newline = '\0';
+    jsonrpc_run(conn->rpc);
+    error = jsonrpc_get_status(conn->rpc);
+    if (error || jsonrpc_get_backlog(conn->rpc)) {
+        return error;
+    }
 
-            process_command(conn, command);
+    for (i = 0; i < 10; i++) {
+        struct jsonrpc_msg *msg;
 
-            ofpbuf_pull(&conn->in, n);
-            if (!conn->in.size) {
-                ofpbuf_clear(&conn->in);
-            }
-            return 0;
+        if (error || conn->request_id) {
+            break;
         }
 
-        ofpbuf_prealloc_tailroom(&conn->in, 128);
-        error = read_fully(conn->fd, ofpbuf_tail(&conn->in),
-                           ofpbuf_tailroom(&conn->in), &bytes_read);
-        conn->in.size += bytes_read;
-        if (conn->in.size > 65536) {
-            VLOG_WARN_RL(&rl, "excess command length, killing connection");
-            return EPROTO;
-        }
-        if (error) {
-            if (error == EAGAIN || error == EWOULDBLOCK) {
-                if (!bytes_read) {
-                    return EAGAIN;
-                }
+        jsonrpc_recv(conn->rpc, &msg);
+        if (msg) {
+            if (msg->type == JSONRPC_REQUEST) {
+                process_command(conn, msg);
             } else {
-                if (error != EOF || conn->in.size != 0) {
-                    VLOG_WARN_RL(&rl, "read failed: %s",
-                                 (error == EOF
-                                  ? "connection dropped mid-command"
-                                  : strerror(error)));
-                }
-                return error;
+                VLOG_WARN_RL(&rl, "%s: received unexpected %s message",
+                             jsonrpc_get_name(conn->rpc),
+                             jsonrpc_msg_type_to_string(msg->type));
+                error = EINVAL;
             }
+            jsonrpc_msg_destroy(msg);
         }
+        error = error ? error : jsonrpc_get_status(conn->rpc);
     }
-}
-
-static int
-run_connection(struct unixctl_conn *conn)
-{
-    int old_state;
-    do {
-        int error;
 
-        old_state = conn->state;
-        switch (conn->state) {
-        case S_RECV:
-            error = run_connection_input(conn);
-            break;
-
-        case S_PROCESS:
-            error = 0;
-            break;
-
-        case S_SEND:
-            error = run_connection_output(conn);
-            break;
-
-        default:
-            NOT_REACHED();
-        }
-        if (error) {
-            return error;
-        }
-    } while (conn->state != old_state);
-    return 0;
+    return error;
 }
 
 static void
 kill_connection(struct unixctl_conn *conn)
 {
     list_remove(&conn->node);
-    ofpbuf_uninit(&conn->in);
-    ds_destroy(&conn->out);
-    close(conn->fd);
+    jsonrpc_close(conn->rpc);
+    json_destroy(conn->request_id);
     free(conn);
 }
 
@@ -460,14 +346,21 @@ unixctl_server_run(struct unixctl_server *server)
     }
 
     for (i = 0; i < 10; i++) {
-        int fd = accept(server->fd, NULL, NULL);
-        if (fd < 0) {
-            if (errno != EAGAIN && errno != EWOULDBLOCK) {
-                VLOG_WARN_RL(&rl, "accept failed: %s", strerror(errno));
-            }
+        struct stream *stream;
+        int error;
+
+        error = pstream_accept(server->listener, &stream);
+        if (!error) {
+            struct unixctl_conn *conn = xzalloc(sizeof *conn);
+            list_push_back(&server->conns, &conn->node);
+            conn->rpc = jsonrpc_open(stream);
+        } else if (error == EAGAIN) {
             break;
+        } else {
+            VLOG_WARN_RL(&rl, "%s: accept failed: %s",
+                         pstream_get_name(server->listener),
+                         strerror(error));
         }
-        new_connection(server, fd);
     }
 
     LIST_FOR_EACH_SAFE (conn, next, node, &server->conns) {
@@ -487,12 +380,11 @@ unixctl_server_wait(struct unixctl_server *server)
         return;
     }
 
-    poll_fd_wait(server->fd, POLLIN);
+    pstream_wait(server->listener);
     LIST_FOR_EACH (conn, node, &server->conns) {
-        if (conn->state == S_RECV) {
-            poll_fd_wait(conn->fd, POLLIN);
-        } else if (conn->state == S_SEND) {
-            poll_fd_wait(conn->fd, POLLOUT);
+        jsonrpc_wait(conn->rpc);
+        if (!jsonrpc_get_backlog(conn->rpc)) {
+            jsonrpc_recv_wait(conn->rpc);
         }
     }
 }
@@ -508,151 +400,92 @@ unixctl_server_destroy(struct unixctl_server *server)
             kill_connection(conn);
         }
 
-        close(server->fd);
-        fatal_signal_unlink_file_now(server->path);
-        free(server->path);
+        pstream_close(server->listener);
         free(server);
     }
 }
 
-/* Connects to a Vlog server socket.  'path' should be the name of a Vlog
- * server socket.  If it does not start with '/', it will be prefixed with
- * the rundir (e.g. /usr/local/var/run/openvswitch).
+/* Connects to a unixctl server socket.  'path' should be the name of a unixctl
+ * server socket.  If it does not start with '/', it will be prefixed with the
+ * rundir (e.g. /usr/local/var/run/openvswitch).
  *
  * Returns 0 if successful, otherwise a positive errno value.  If successful,
- * sets '*clientp' to the new unixctl_client, otherwise to NULL. */
+ * sets '*client' to the new jsonrpc, otherwise to NULL. */
 int
-unixctl_client_create(const char *path, struct unixctl_client **clientp)
+unixctl_client_create(const char *path, struct jsonrpc **client)
 {
-    static int counter;
-    struct unixctl_client *client;
+    char *abs_path, *unix_path;
+    struct stream *stream;
     int error;
-    int fd = -1;
-
-    /* Determine location. */
-    client = xmalloc(sizeof *client);
-    client->connect_path = abs_file_name(ovs_rundir(), path);
-    client->bind_path = xasprintf("/tmp/vlog.%ld.%d",
-                                  (long int) getpid(), counter++);
-
-    /* Open socket. */
-    fd = make_unix_socket(SOCK_STREAM, false, false,
-                          client->bind_path, client->connect_path);
-    if (fd < 0) {
-        error = -fd;
-        goto error;
-    }
 
-    /* Bind socket to stream. */
-    client->stream = fdopen(fd, "r+");
-    if (!client->stream) {
-        error = errno;
-        VLOG_WARN("%s: fdopen failed (%s)",
-                  client->connect_path, strerror(error));
-        goto error;
-    }
-    *clientp = client;
-    return 0;
+    *client = NULL;
 
-error:
-    if (fd >= 0) {
-        close(fd);
-    }
-    free(client->connect_path);
-    free(client->bind_path);
-    free(client);
-    *clientp = NULL;
-    return error;
-}
+    abs_path = abs_file_name(ovs_rundir(), path);
+    unix_path = xasprintf("unix:%s", abs_path);
+    error = stream_open_block(stream_open(unix_path, &stream), &stream);
+    free(unix_path);
+    free(abs_path);
 
-/* Destroys 'client'. */
-void
-unixctl_client_destroy(struct unixctl_client *client)
-{
-    if (client) {
-        fatal_signal_unlink_file_now(client->bind_path);
-        free(client->bind_path);
-        free(client->connect_path);
-        fclose(client->stream);
-        free(client);
+    if (error) {
+        VLOG_WARN("failed to connect to %s", path);
+        return error;
     }
+
+    *client = jsonrpc_open(stream);
+    return 0;
 }
 
-/* Sends 'request' to the server socket and waits for a reply.  Returns 0 if
- * successful, otherwise to a positive errno value.  If successful, sets
- * '*reply' to the reply, which the caller must free, otherwise to NULL. */
+/* Executes 'command' on the server with an argument vector 'argv' containing
+ * 'argc' elements.  If successfully communicated with the server, returns 0
+ * and sets '*result', or '*err' (not both) to the result or error the server
+ * returned.  Otherwise, sets '*result' and '*err' to NULL and returns a
+ * positive errno value.  The caller is responsible for freeing '*result' or
+ * '*err' if not NULL. */
 int
-unixctl_client_transact(struct unixctl_client *client,
-                        const char *request,
-                        int *reply_code, char **reply_body)
+unixctl_client_transact(struct jsonrpc *client, const char *command, int argc,
+                        char *argv[], char **result, char **err)
 {
-    struct ds line = DS_EMPTY_INITIALIZER;
-    struct ds reply = DS_EMPTY_INITIALIZER;
-    int error;
+    struct jsonrpc_msg *request, *reply;
+    struct json **json_args, *params;
+    int error, i;
 
-    /* Send 'request' to server.  Add a new-line if 'request' didn't end in
-     * one. */
-    fputs(request, client->stream);
-    if (request[0] == '\0' || request[strlen(request) - 1] != '\n') {
-        putc('\n', client->stream);
+    *result = NULL;
+    *err = NULL;
+
+    json_args = xmalloc(argc * sizeof *json_args);
+    for (i = 0; i < argc; i++) {
+        json_args[i] = json_string_create(argv[i]);
     }
-    if (ferror(client->stream)) {
-        VLOG_WARN("error sending request to %s: %s",
-                  client->connect_path, strerror(errno));
-        return errno;
+    params = json_array_create(json_args, argc);
+    request = jsonrpc_create_request(command, params, NULL);
+
+    error = jsonrpc_transact_block(client, request, &reply);
+    if (error) {
+        VLOG_WARN("error communicating with %s: %s", jsonrpc_get_name(client),
+                  strerror(error));
+        return error;
     }
 
-    /* Wait for response. */
-    *reply_code = -1;
-    for (;;) {
-        const char *s;
-
-        error = ds_get_line(&line, client->stream);
-        if (error) {
-            VLOG_WARN("error reading reply from %s: %s",
-                      client->connect_path,
-                      ovs_retval_to_string(error));
-            goto error;
+    if (reply->error) {
+        if (reply->error->type == JSON_STRING) {
+            *err = xstrdup(json_string(reply->error));
+        } else {
+            VLOG_WARN("%s: unexpected error type in JSON RPC reply: %s",
+                      jsonrpc_get_name(client),
+                      json_type_to_string(reply->error->type));
+            error = EINVAL;
         }
-
-        s = ds_cstr(&line);
-        if (*reply_code == -1) {
-            if (!isdigit((unsigned char)s[0])
-                    || !isdigit((unsigned char)s[1])
-                    || !isdigit((unsigned char)s[2])) {
-                VLOG_WARN("reply from %s does not start with 3-digit code",
-                          client->connect_path);
-                error = EPROTO;
-                goto error;
-            }
-            sscanf(s, "%3d", reply_code);
+    } else if (reply->result) {
+        if (reply->result->type == JSON_STRING) {
+            *result = xstrdup(json_string(reply->result));
         } else {
-            if (s[0] == '.') {
-                if (s[1] == '\0') {
-                    break;
-                }
-                s++;
-            }
-            ds_put_cstr(&reply, s);
-            ds_put_char(&reply, '\n');
+            VLOG_WARN("%s: unexpected result type in JSON rpc reply: %s",
+                      jsonrpc_get_name(client),
+                      json_type_to_string(reply->result->type));
+            error = EINVAL;
         }
     }
-    *reply_body = ds_cstr(&reply);
-    ds_destroy(&line);
-    return 0;
 
-error:
-    ds_destroy(&line);
-    ds_destroy(&reply);
-    *reply_code = 0;
-    *reply_body = NULL;
-    return error == EOF ? EPROTO : error;
-}
-
-/* Returns the path of the server socket to which 'client' is connected.  The
- * caller must not modify or free the returned string. */
-const char *
-unixctl_client_target(const struct unixctl_client *client)
-{
-    return client->connect_path;
+    jsonrpc_msg_destroy(reply);
+    return error;
 }
diff --git a/lib/unixctl.h b/lib/unixctl.h
index 8eb190b0..13a8096e 100644
--- a/lib/unixctl.h
+++ b/lib/unixctl.h
@@ -29,13 +29,12 @@ void unixctl_server_wait(struct unixctl_server *);
 void unixctl_server_destroy(struct unixctl_server *);
 
 /* Client for Unix domain socket control connection. */
-struct unixctl_client;
-int unixctl_client_create(const char *path, struct unixctl_client **);
-void unixctl_client_destroy(struct unixctl_client *);
-int unixctl_client_transact(struct unixctl_client *,
-                            const char *request,
-                            int *reply_code, char **reply_body);
-const char *unixctl_client_target(const struct unixctl_client *);
+struct jsonrpc;
+int unixctl_client_create(const char *path, struct jsonrpc **client);
+int unixctl_client_transact(struct jsonrpc *client,
+                            const char *command,
+                            int argc, char *argv[],
+                            char **result, char **error);
 
 /* Command registration. */
 struct unixctl_conn;
@@ -44,8 +43,8 @@ typedef void unixctl_cb_func(struct unixctl_conn *,
 void unixctl_command_register(const char *name, const char *usage,
                               int min_args, int max_args,
                               unixctl_cb_func *cb, void *aux);
-void unixctl_command_reply(struct unixctl_conn *, int code,
-                           const char *body);
+void unixctl_command_reply_error(struct unixctl_conn *, const char *error);
+void unixctl_command_reply(struct unixctl_conn *, const char *body);
 
 #ifdef  __cplusplus
 }
diff --git a/lib/vlog.c b/lib/vlog.c
index 07ba5e02..4c97ef40 100644
--- a/lib/vlog.c
+++ b/lib/vlog.c
@@ -439,12 +439,12 @@ vlog_unixctl_set(struct unixctl_conn *conn, int argc, const char *argv[],
     for (i = 1; i < argc; i++) {
         char *msg = vlog_set_levels_from_string(argv[i]);
         if (msg) {
-            unixctl_command_reply(conn, 501, msg);
+            unixctl_command_reply_error(conn, msg);
             free(msg);
             return;
         }
     }
-    unixctl_command_reply(conn, 202, NULL);
+    unixctl_command_reply(conn, NULL);
 }
 
 static void
@@ -452,7 +452,7 @@ vlog_unixctl_list(struct unixctl_conn *conn, int argc OVS_UNUSED,
                   const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
 {
     char *msg = vlog_get_levels();
-    unixctl_command_reply(conn, 200, msg);
+    unixctl_command_reply(conn, msg);
     free(msg);
 }
 
@@ -463,12 +463,12 @@ vlog_unixctl_reopen(struct unixctl_conn *conn, int argc OVS_UNUSED,
     if (log_file_name) {
         int error = vlog_reopen_log_file();
         if (error) {
-            unixctl_command_reply(conn, 503, strerror(errno));
+            unixctl_command_reply_error(conn, strerror(errno));
         } else {
-            unixctl_command_reply(conn, 202, NULL);
+            unixctl_command_reply(conn, NULL);
         }
     } else {
-        unixctl_command_reply(conn, 403, "Logging to file not configured");
+        unixctl_command_reply_error(conn, "Logging to file not configured");
     }
 }
 
diff --git a/ofproto/ofproto-dpif.c b/ofproto/ofproto-dpif.c
index feb1342e..d8d1e384 100644
--- a/ofproto/ofproto-dpif.c
+++ b/ofproto/ofproto-dpif.c
@@ -5898,7 +5898,7 @@ ofproto_unixctl_fdb_flush(struct unixctl_conn *conn, int argc,
     if (argc > 1) {
         ofproto = ofproto_dpif_lookup(argv[1]);
         if (!ofproto) {
-            unixctl_command_reply(conn, 501, "no such bridge");
+            unixctl_command_reply_error(conn, "no such bridge");
             return;
         }
         mac_learning_flush(ofproto->ml, &ofproto->revalidate_set);
@@ -5908,7 +5908,7 @@ ofproto_unixctl_fdb_flush(struct unixctl_conn *conn, int argc,
         }
     }
 
-    unixctl_command_reply(conn, 200, "table successfully flushed");
+    unixctl_command_reply(conn, "table successfully flushed");
 }
 
 static void
@@ -5921,7 +5921,7 @@ ofproto_unixctl_fdb_show(struct unixctl_conn *conn, int argc OVS_UNUSED,
 
     ofproto = ofproto_dpif_lookup(argv[1]);
     if (!ofproto) {
-        unixctl_command_reply(conn, 501, "no such bridge");
+        unixctl_command_reply_error(conn, "no such bridge");
         return;
     }
 
@@ -5933,7 +5933,7 @@ ofproto_unixctl_fdb_show(struct unixctl_conn *conn, int argc OVS_UNUSED,
                       e->vlan, ETH_ADDR_ARGS(e->mac),
                       mac_entry_age(ofproto->ml, e));
     }
-    unixctl_command_reply(conn, 200, ds_cstr(&ds));
+    unixctl_command_reply(conn, ds_cstr(&ds));
     ds_destroy(&ds);
 }
 
@@ -6038,8 +6038,8 @@ ofproto_unixctl_trace(struct unixctl_conn *conn, int argc, const char *argv[],
 
     ofproto = ofproto_dpif_lookup(dpname);
     if (!ofproto) {
-        unixctl_command_reply(conn, 501, "Unknown ofproto (use ofproto/list "
-                              "for help)");
+        unixctl_command_reply_error(conn, "Unknown ofproto (use ofproto/list "
+                                    "for help)");
         goto exit;
     }
     if (argc == 3 || (argc == 4 && !strcmp(argv[3], "-generate"))) {
@@ -6052,7 +6052,7 @@ ofproto_unixctl_trace(struct unixctl_conn *conn, int argc, const char *argv[],
         ofpbuf_init(&odp_key, 0);
         error = odp_flow_key_from_string(flow_s, NULL, &odp_key);
         if (error) {
-            unixctl_command_reply(conn, 501, "Bad flow syntax");
+            unixctl_command_reply_error(conn, "Bad flow syntax");
             goto exit;
         }
 
@@ -6061,7 +6061,7 @@ ofproto_unixctl_trace(struct unixctl_conn *conn, int argc, const char *argv[],
                                               odp_key.size, &flow,
                                               &initial_tci, NULL);
         if (error == ODP_FIT_ERROR) {
-            unixctl_command_reply(conn, 501, "Invalid flow");
+            unixctl_command_reply_error(conn, "Invalid flow");
             goto exit;
         }
 
@@ -6083,7 +6083,7 @@ ofproto_unixctl_trace(struct unixctl_conn *conn, int argc, const char *argv[],
 
         msg = eth_from_hex(packet_s, &packet);
         if (msg) {
-            unixctl_command_reply(conn, 501, msg);
+            unixctl_command_reply_error(conn, msg);
             goto exit;
         }
 
@@ -6095,7 +6095,7 @@ ofproto_unixctl_trace(struct unixctl_conn *conn, int argc, const char *argv[],
         flow_extract(packet, priority, tun_id, in_port, &flow);
         initial_tci = flow.vlan_tci;
     } else {
-        unixctl_command_reply(conn, 501, "Bad command syntax");
+        unixctl_command_reply_error(conn, "Bad command syntax");
         goto exit;
     }
 
@@ -6135,7 +6135,7 @@ ofproto_unixctl_trace(struct unixctl_conn *conn, int argc, const char *argv[],
         }
     }
 
-    unixctl_command_reply(conn, 200, ds_cstr(&result));
+    unixctl_command_reply(conn, ds_cstr(&result));
 
 exit:
     ds_destroy(&result);
@@ -6148,7 +6148,7 @@ ofproto_dpif_clog(struct unixctl_conn *conn OVS_UNUSED, int argc OVS_UNUSED,
                   const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
 {
     clogged = true;
-    unixctl_command_reply(conn, 200, NULL);
+    unixctl_command_reply(conn, NULL);
 }
 
 static void
@@ -6156,7 +6156,7 @@ ofproto_dpif_unclog(struct unixctl_conn *conn OVS_UNUSED, int argc OVS_UNUSED,
                     const char *argv[] OVS_UNUSED, void *aux OVS_UNUSED)
 {
     clogged = false;
-    unixctl_command_reply(conn, 200, NULL);
+    unixctl_command_reply(conn, NULL);
 }
 
 /* Runs a self-check of flow translations in 'ofproto'.  Appends a message to
@@ -6195,8 +6195,8 @@ ofproto_dpif_self_check(struct unixctl_conn *conn,
     if (argc > 1) {
         ofproto = ofproto_dpif_lookup(argv[1]);
         if (!ofproto) {
-            unixctl_command_reply(conn, 501, "Unknown ofproto (use "
-                                  "ofproto/list for help)");
+            unixctl_command_reply_error(conn, "Unknown ofproto (use "
+                                        "ofproto/list for help)");
             return;
         }
         ofproto_dpif_self_check__(ofproto, &reply);
@@ -6206,7 +6206,7 @@ ofproto_dpif_self_check(struct unixctl_conn *conn,
         }
     }
 
-    unixctl_command_reply(conn, 200, ds_cstr(&reply));
+    unixctl_command_reply(conn, ds_cstr(&reply));
     ds_destroy(&reply);
 }
 
diff --git a/ofproto/ofproto.c b/ofproto/ofproto.c
index 08a18a46..ce36d956 100644
--- a/ofproto/ofproto.c
+++ b/ofproto/ofproto.c
@@ -3983,7 +3983,7 @@ ofproto_unixctl_list(struct unixctl_conn *conn, int argc OVS_UNUSED,
     HMAP_FOR_EACH (ofproto, hmap_node, &all_ofprotos) {
         ds_put_format(&results, "%s\n", ofproto->name);
     }
-    unixctl_command_reply(conn, 200, ds_cstr(&results));
+    unixctl_command_reply(conn, ds_cstr(&results));
     ds_destroy(&results);
 }
 
diff --git a/ovsdb/automake.mk b/ovsdb/automake.mk
index 3840ffce..d2e3f9ad 100644
--- a/ovsdb/automake.mk
+++ b/ovsdb/automake.mk
@@ -36,7 +36,7 @@ MAN_FRAGMENTS += \
 # ovsdb-tool
 bin_PROGRAMS += ovsdb/ovsdb-tool
 ovsdb_ovsdb_tool_SOURCES = ovsdb/ovsdb-tool.c
-ovsdb_ovsdb_tool_LDADD = ovsdb/libovsdb.a lib/libopenvswitch.a
+ovsdb_ovsdb_tool_LDADD = ovsdb/libovsdb.a lib/libopenvswitch.a $(SSL_LIBS)
 # ovsdb-tool.1
 man_MANS += ovsdb/ovsdb-tool.1
 DISTCLEANFILES += ovsdb/ovsdb-tool.1
diff --git a/ovsdb/ovsdb-server.c b/ovsdb/ovsdb-server.c
index 41238fbb..9e0636e9 100644
--- a/ovsdb/ovsdb-server.c
+++ b/ovsdb/ovsdb-server.c
@@ -618,7 +618,7 @@ ovsdb_server_exit(struct unixctl_conn *conn, int argc OVS_UNUSED,
 {
     bool *exiting = exiting_;
     *exiting = true;
-    unixctl_command_reply(conn, 200, NULL);
+    unixctl_command_reply(conn, NULL);
 }
 
 static void
@@ -631,11 +631,11 @@ ovsdb_server_compact(struct unixctl_conn *conn, int argc OVS_UNUSED,
     VLOG_INFO("compacting database by user request");
     error = ovsdb_file_compact(file);
     if (!error) {
-        unixctl_command_reply(conn, 200, NULL);
+        unixctl_command_reply(conn, NULL);
     } else {
         char *s = ovsdb_error_to_string(error);
         ovsdb_error_destroy(error);
-        unixctl_command_reply(conn, 503, s);
+        unixctl_command_reply_error(conn, s);
         free(s);
     }
 }
@@ -649,7 +649,7 @@ ovsdb_server_reconnect(struct unixctl_conn *conn, int argc OVS_UNUSED,
     struct ovsdb_jsonrpc_server *jsonrpc = jsonrpc_;
 
     ovsdb_jsonrpc_server_reconnect(jsonrpc);
-    unixctl_command_reply(conn, 200, NULL);
+    unixctl_command_reply(conn, NULL);
 }
 
 static void
diff --git a/tests/automake.mk b/tests/automake.mk
index 28b2749a..b133467c 100644
--- a/tests/automake.mk
+++ b/tests/automake.mk
@@ -203,27 +203,27 @@ $(srcdir)/package.m4: $(top_srcdir)/configure.ac
 
 noinst_PROGRAMS += tests/test-aes128
 tests_test_aes128_SOURCES = tests/test-aes128.c
-tests_test_aes128_LDADD = lib/libopenvswitch.a
+tests_test_aes128_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-bundle
 tests_test_bundle_SOURCES = tests/test-bundle.c
-tests_test_bundle_LDADD = lib/libopenvswitch.a
+tests_test_bundle_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-classifier
 tests_test_classifier_SOURCES = tests/test-classifier.c
-tests_test_classifier_LDADD = lib/libopenvswitch.a
+tests_test_classifier_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-csum
 tests_test_csum_SOURCES = tests/test-csum.c
-tests_test_csum_LDADD = lib/libopenvswitch.a
+tests_test_csum_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-file_name
 tests_test_file_name_SOURCES = tests/test-file_name.c
-tests_test_file_name_LDADD = lib/libopenvswitch.a
+tests_test_file_name_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-flows
 tests_test_flows_SOURCES = tests/test-flows.c
-tests_test_flows_LDADD = lib/libopenvswitch.a
+tests_test_flows_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 dist_check_SCRIPTS = tests/flowgen.pl
 
 noinst_PROGRAMS += tests/test-hash
@@ -232,15 +232,15 @@ tests_test_hash_LDADD = lib/libopenvswitch.a
 
 noinst_PROGRAMS += tests/test-heap
 tests_test_heap_SOURCES = tests/test-heap.c
-tests_test_heap_LDADD = lib/libopenvswitch.a
+tests_test_heap_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-hmap
 tests_test_hmap_SOURCES = tests/test-hmap.c
-tests_test_hmap_LDADD = lib/libopenvswitch.a
+tests_test_hmap_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-json
 tests_test_json_SOURCES = tests/test-json.c
-tests_test_json_LDADD = lib/libopenvswitch.a
+tests_test_json_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-jsonrpc
 tests_test_jsonrpc_SOURCES = tests/test-jsonrpc.c
@@ -252,35 +252,35 @@ tests_test_list_LDADD = lib/libopenvswitch.a
 
 noinst_PROGRAMS += tests/test-lockfile
 tests_test_lockfile_SOURCES = tests/test-lockfile.c
-tests_test_lockfile_LDADD = lib/libopenvswitch.a
+tests_test_lockfile_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-multipath
 tests_test_multipath_SOURCES = tests/test-multipath.c
-tests_test_multipath_LDADD = lib/libopenvswitch.a
+tests_test_multipath_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-packets
 tests_test_packets_SOURCES = tests/test-packets.c
-tests_test_packets_LDADD = lib/libopenvswitch.a
+tests_test_packets_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-random
 tests_test_random_SOURCES = tests/test-random.c
-tests_test_random_LDADD = lib/libopenvswitch.a
+tests_test_random_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-stp
 tests_test_stp_SOURCES = tests/test-stp.c
-tests_test_stp_LDADD = lib/libopenvswitch.a
+tests_test_stp_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-netflow
 tests_test_netflow_SOURCES = tests/test-netflow.c
-tests_test_netflow_LDADD = lib/libopenvswitch.a
+tests_test_netflow_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-unix-socket
 tests_test_unix_socket_SOURCES = tests/test-unix-socket.c
-tests_test_unix_socket_LDADD = lib/libopenvswitch.a
+tests_test_unix_socket_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-odp
 tests_test_odp_SOURCES = tests/test-odp.c
-tests_test_odp_LDADD = lib/libopenvswitch.a
+tests_test_odp_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-ovsdb
 tests_test_ovsdb_SOURCES = \
@@ -302,15 +302,15 @@ tests/idltest.c: tests/idltest.h
 
 noinst_PROGRAMS += tests/test-reconnect
 tests_test_reconnect_SOURCES = tests/test-reconnect.c
-tests_test_reconnect_LDADD = lib/libopenvswitch.a
+tests_test_reconnect_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-sha1
 tests_test_sha1_SOURCES = tests/test-sha1.c
-tests_test_sha1_LDADD = lib/libopenvswitch.a
+tests_test_sha1_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-timeval
 tests_test_timeval_SOURCES = tests/test-timeval.c
-tests_test_timeval_LDADD = lib/libopenvswitch.a
+tests_test_timeval_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-strtok_r
 tests_test_strtok_r_SOURCES = tests/test-strtok_r.c
@@ -320,11 +320,11 @@ tests_test_type_props_SOURCES = tests/test-type-props.c
 
 noinst_PROGRAMS += tests/test-util
 tests_test_util_SOURCES = tests/test-util.c
-tests_test_util_LDADD = lib/libopenvswitch.a
+tests_test_util_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-uuid
 tests_test_uuid_SOURCES = tests/test-uuid.c
-tests_test_uuid_LDADD = lib/libopenvswitch.a
+tests_test_uuid_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += tests/test-vconn
 tests_test_vconn_SOURCES = tests/test-vconn.c
diff --git a/tests/test-netflow.c b/tests/test-netflow.c
index b85c663e..8b474ea9 100644
--- a/tests/test-netflow.c
+++ b/tests/test-netflow.c
@@ -312,5 +312,5 @@ test_netflow_exit(struct unixctl_conn *conn,
 {
     bool *exiting = exiting_;
     *exiting = true;
-    unixctl_command_reply(conn, 200, "");
+    unixctl_command_reply(conn, NULL);
 }
diff --git a/utilities/automake.mk b/utilities/automake.mk
index 89078cb8..3e66aee1 100644
--- a/utilities/automake.mk
+++ b/utilities/automake.mk
@@ -83,13 +83,13 @@ man_MANS += \
 dist_man_MANS += utilities/ovs-ctl.8
 
 utilities_ovs_appctl_SOURCES = utilities/ovs-appctl.c
-utilities_ovs_appctl_LDADD = lib/libopenvswitch.a
+utilities_ovs_appctl_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 utilities_ovs_controller_SOURCES = utilities/ovs-controller.c
 utilities_ovs_controller_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 utilities_ovs_dpctl_SOURCES = utilities/ovs-dpctl.c
-utilities_ovs_dpctl_LDADD = lib/libopenvswitch.a
+utilities_ovs_dpctl_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 utilities_ovs_ofctl_SOURCES = utilities/ovs-ofctl.c
 utilities_ovs_ofctl_LDADD = \
@@ -103,15 +103,15 @@ utilities_ovs_vsctl_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 if HAVE_NETLINK
 sbin_PROGRAMS += utilities/ovs-vlan-bug-workaround
 utilities_ovs_vlan_bug_workaround_SOURCES = utilities/ovs-vlan-bug-workaround.c
-utilities_ovs_vlan_bug_workaround_LDADD = lib/libopenvswitch.a
+utilities_ovs_vlan_bug_workaround_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 noinst_PROGRAMS += utilities/nlmon
 utilities_nlmon_SOURCES = utilities/nlmon.c
-utilities_nlmon_LDADD = lib/libopenvswitch.a
+utilities_nlmon_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 endif
 
 bin_PROGRAMS += utilities/ovs-benchmark
 utilities_ovs_benchmark_SOURCES = utilities/ovs-benchmark.c
-utilities_ovs_benchmark_LDADD = lib/libopenvswitch.a
+utilities_ovs_benchmark_LDADD = lib/libopenvswitch.a $(SSL_LIBS)
 
 include utilities/bugtool/automake.mk
diff --git a/utilities/ovs-appctl.c b/utilities/ovs-appctl.c
index ebfd54a1..241b6c04 100644
--- a/utilities/ovs-appctl.c
+++ b/utilities/ovs-appctl.c
@@ -26,6 +26,7 @@
 #include "daemon.h"
 #include "dirs.h"
 #include "dynamic-string.h"
+#include "jsonrpc.h"
 #include "process.h"
 #include "timeval.h"
 #include "unixctl.h"
@@ -33,16 +34,17 @@
 
 static void usage(void);
 static const char *parse_command_line(int argc, char *argv[]);
-static struct unixctl_client *connect_to_target(const char *target);
+static struct jsonrpc *connect_to_target(const char *target);
 
 int
 main(int argc, char *argv[])
 {
-    struct unixctl_client *client;
+    char *cmd_result, *cmd_error;
+    struct jsonrpc *client;
+    char *cmd, **cmd_argv;
     const char *target;
-    int code, error;
-    char *request;
-    char *reply;
+    int cmd_argc;
+    int error;
 
     set_program_name(argv[0]);
 
@@ -51,22 +53,28 @@ main(int argc, char *argv[])
     client = connect_to_target(target);
 
     /* Transact request and process reply. */
-    request = process_escape_args(argv + optind);
-    error = unixctl_client_transact(client, request, &code, &reply);
-    free(request);
+    cmd = argv[optind++];
+    cmd_argc = argc - optind;
+    cmd_argv = cmd_argc ? argv + optind : NULL;
+    error = unixctl_client_transact(client, cmd, cmd_argc, cmd_argv,
+                                    &cmd_result, &cmd_error);
     if (error) {
         ovs_fatal(error, "%s: transaction error", target);
     }
-    if (code / 100 != 2) {
-        fputs(reply, stderr);
-        ovs_error(0, "%s: server returned reply code %03d", target, code);
+
+    if (cmd_error) {
+        fputs(cmd_error, stderr);
+        ovs_error(0, "%s: server returned an error", target);
         exit(2);
+    } else if (cmd_result) {
+        fputs(cmd_result, stdout);
+    } else {
+        NOT_REACHED();
     }
-    fputs(reply, stdout);
-
-    unixctl_client_destroy(client);
-    free(reply);
 
+    jsonrpc_close(client);
+    free(cmd_result);
+    free(cmd_error);
     return 0;
 }
 
@@ -159,10 +167,10 @@ parse_command_line(int argc, char *argv[])
     return target ? target : "ovs-vswitchd";
 }
 
-static struct unixctl_client *
+static struct jsonrpc *
 connect_to_target(const char *target)
 {
-    struct unixctl_client *client;
+    struct jsonrpc *client;
     char *socket_name;
     int error;
 
diff --git a/utilities/ovs-ofctl.c b/utilities/ovs-ofctl.c
index bf7a2c12..fd0829ea 100644
--- a/utilities/ovs-ofctl.c
+++ b/utilities/ovs-ofctl.c
@@ -254,7 +254,7 @@ ofctl_exit(struct unixctl_conn *conn, int argc OVS_UNUSED,
 {
     bool *exiting = exiting_;
     *exiting = true;
-    unixctl_command_reply(conn, 200, "");
+    unixctl_command_reply(conn, NULL);
 }
 
 static void run(int retval, const char *message, ...)
@@ -910,7 +910,12 @@ ofctl_send(struct unixctl_conn *conn, int argc,
             ds_put_cstr(&reply, "sent\n");
         }
     }
-    unixctl_command_reply(conn, ok ? 200 : 501, ds_cstr(&reply));
+
+    if (ok) {
+        unixctl_command_reply(conn, ds_cstr(&reply));
+    } else {
+        unixctl_command_reply_error(conn, ds_cstr(&reply));
+    }
     ds_destroy(&reply);
 }
 
@@ -928,7 +933,7 @@ ofctl_barrier(struct unixctl_conn *conn, int argc OVS_UNUSED,
     int error;
 
     if (aux->conn) {
-        unixctl_command_reply(conn, 501, "already waiting for barrier reply");
+        unixctl_command_reply_error(conn, "already waiting for barrier reply");
         return;
     }
 
@@ -939,7 +944,7 @@ ofctl_barrier(struct unixctl_conn *conn, int argc OVS_UNUSED,
     error = vconn_send_block(aux->vconn, msg);
     if (error) {
         ofpbuf_delete(msg);
-        unixctl_command_reply(conn, 501, strerror(error));
+        unixctl_command_reply_error(conn, strerror(error));
     } else {
         aux->conn = conn;
     }
@@ -953,14 +958,14 @@ ofctl_set_output_file(struct unixctl_conn *conn, int argc OVS_UNUSED,
 
     fd = open(argv[1], O_CREAT | O_TRUNC | O_WRONLY, 0666);
     if (fd < 0) {
-        unixctl_command_reply(conn, 501, strerror(errno));
+        unixctl_command_reply_error(conn, strerror(errno));
         return;
     }
 
     fflush(stderr);
     dup2(fd, STDERR_FILENO);
     close(fd);
-    unixctl_command_reply(conn, 200, "");
+    unixctl_command_reply(conn, NULL);
 }
 
 static void
@@ -1013,7 +1018,7 @@ monitor_vconn(struct vconn *vconn)
             ofpbuf_delete(b);
 
             if (barrier_aux.conn && msg_type == OFPT_BARRIER_REPLY) {
-                unixctl_command_reply(barrier_aux.conn, 200, "");
+                unixctl_command_reply(barrier_aux.conn, NULL);
                 barrier_aux.conn = NULL;
             }
         }
diff --git a/vswitchd/bridge.c b/vswitchd/bridge.c
index e857077a..5b0f66fe 100644
--- a/vswitchd/bridge.c
+++ b/vswitchd/bridge.c
@@ -2084,7 +2084,7 @@ qos_unixctl_show(struct unixctl_conn *conn, int argc OVS_UNUSED,
 
     iface = iface_find(argv[1]);
     if (!iface) {
-        unixctl_command_reply(conn, 501, "no such interface");
+        unixctl_command_reply_error(conn, "no such interface");
         return;
     }
 
@@ -2104,10 +2104,10 @@ qos_unixctl_show(struct unixctl_conn *conn, int argc OVS_UNUSED,
         if (error) {
             ds_put_format(&ds, "failed to dump queues: %s", strerror(error));
         }
-        unixctl_command_reply(conn, 200, ds_cstr(&ds));
+        unixctl_command_reply(conn, ds_cstr(&ds));
     } else {
         ds_put_format(&ds, "QoS not configured on %s\n", iface->name);
-        unixctl_command_reply(conn, 501, ds_cstr(&ds));
+        unixctl_command_reply_error(conn, ds_cstr(&ds));
     }
 
     shash_destroy_free_data(&sh);
@@ -2189,14 +2189,14 @@ bridge_unixctl_dump_flows(struct unixctl_conn *conn, int argc OVS_UNUSED,
 
     br = bridge_lookup(argv[1]);
     if (!br) {
-        unixctl_command_reply(conn, 501, "Unknown bridge");
+        unixctl_command_reply_error(conn, "Unknown bridge");
         return;
     }
 
     ds_init(&results);
     ofproto_get_all_flows(br->ofproto, &results);
 
-    unixctl_command_reply(conn, 200, ds_cstr(&results));
+    unixctl_command_reply(conn, ds_cstr(&results));
     ds_destroy(&results);
 }
 
@@ -2211,7 +2211,7 @@ bridge_unixctl_reconnect(struct unixctl_conn *conn, int argc,
     if (argc > 1) {
         br = bridge_lookup(argv[1]);
         if (!br) {
-            unixctl_command_reply(conn, 501, "Unknown bridge");
+            unixctl_command_reply_error(conn,  "Unknown bridge");
             return;
         }
         ofproto_reconnect_controllers(br->ofproto);
@@ -2220,7 +2220,7 @@ bridge_unixctl_reconnect(struct unixctl_conn *conn, int argc,
             ofproto_reconnect_controllers(br->ofproto);
         }
     }
-    unixctl_command_reply(conn, 200, NULL);
+    unixctl_command_reply(conn, NULL);
 }
 
 static size_t
diff --git a/vswitchd/ovs-vswitchd.c b/vswitchd/ovs-vswitchd.c
index 2360086c..76dc52bb 100644
--- a/vswitchd/ovs-vswitchd.c
+++ b/vswitchd/ovs-vswitchd.c
@@ -247,5 +247,5 @@ ovs_vswitchd_exit(struct unixctl_conn *conn, int argc OVS_UNUSED,
 {
     bool *exiting = exiting_;
     *exiting = true;
-    unixctl_command_reply(conn, 200, NULL);
+    unixctl_command_reply(conn, NULL);
 }