ovsdb: Add replication support and refactor files in terms of replication.

An upcoming commit will add support for replicating tables across JSON-RPC
connection.  As a prerequisite ovsdb itself must support basic replication.
This commit adds that support and then reimplements the ovsdb file storage
in terms of that replication.

10 files changed, 461 insertions, 302 deletions

diff --git a/ovsdb/automake.mk b/ovsdb/automake.mk
index b996a306f..fbc781bca 100644
--- a/ovsdb/automake.mk
+++ b/ovsdb/automake.mk
@@ -6,6 +6,8 @@ ovsdb_libovsdb_a_SOURCES = \
 	ovsdb/condition.c \
 	ovsdb/condition.h \
 	ovsdb/execution.c \
+	ovsdb/file.c \
+	ovsdb/file.h \
 	ovsdb/jsonrpc-server.c \
 	ovsdb/jsonrpc-server.h \
 	ovsdb/log.c \
diff --git a/ovsdb/execution.c b/ovsdb/execution.c
index 6bad67c2b..9edc1a971 100644
--- a/ovsdb/execution.c
+++ b/ovsdb/execution.c
@@ -20,8 +20,8 @@
 
 #include "column.h"
 #include "condition.h"
+#include "file.h"
 #include "json.h"
-#include "log.h"
 #include "ovsdb-data.h"
 #include "ovsdb-error.h"
 #include "ovsdb-parser.h"
@@ -47,7 +47,6 @@ typedef struct ovsdb_error *ovsdb_operation_executor(struct ovsdb_execution *,
                                                      struct ovsdb_parser *,
                                                      struct json *result);
 
-static struct ovsdb_error *do_commit(struct ovsdb_execution *);
 static ovsdb_operation_executor ovsdb_execute_insert;
 static ovsdb_operation_executor ovsdb_execute_select;
 static ovsdb_operation_executor ovsdb_execute_update;
@@ -168,8 +167,7 @@ ovsdb_execute(struct ovsdb *db, const struct json *params,
     }
 
     if (!error) {
-        /* Commit transaction.  Bail if commit encounters error.  */
-        error = do_commit(&x);
+        error = ovsdb_txn_commit(x.txn, x.durable);
         if (error) {
             json_array_add(results, ovsdb_error_to_json(error));
         }
@@ -208,38 +206,6 @@ ovsdb_execute_abort(struct ovsdb_execution *x UNUSED,
     return ovsdb_error("aborted", "aborted by request");
 }
 
-static struct ovsdb_error *
-do_commit(struct ovsdb_execution *x)
-{
-    if (x->db->log) {
-        struct ovsdb_error *error;
-        struct json *json;
-
-        json = ovsdb_txn_to_json(x->txn);
-        if (!json) {
-            /* Nothing to commit. */
-            return NULL;
-        }
-
-        error = ovsdb_log_write(x->db->log, json);
-        json_destroy(json);
-        if (error) {
-            return ovsdb_wrap_error(error, "writing transaction failed");
-        }
-
-        if (x->durable) {
-            error = ovsdb_log_commit(x->db->log);
-            if (error) {
-                return ovsdb_wrap_error(error,
-                                        "committing transaction failed");
-            }
-        }
-    }
-
-    ovsdb_txn_commit(x->txn);
-    return NULL;
-}
-
 static struct ovsdb_table *
 parse_table(struct ovsdb_execution *x,
             struct ovsdb_parser *parser, const char *member)
diff --git a/ovsdb/file.c b/ovsdb/file.c
new file mode 100644
index 000000000..377ca280f
--- /dev/null
+++ b/ovsdb/file.c
@@ -0,0 +1,341 @@
+/* Copyright (c) 2009 Nicira Networks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <config.h>
+
+#include "file.h"
+
+#include <assert.h>
+#include <fcntl.h>
+
+#include "column.h"
+#include "log.h"
+#include "json.h"
+#include "ovsdb.h"
+#include "ovsdb-error.h"
+#include "row.h"
+#include "table.h"
+#include "transaction.h"
+#include "uuid.h"
+#include "util.h"
+
+#define THIS_MODULE VLM_ovsdb_file
+#include "vlog.h"
+
+static struct ovsdb_error *ovsdb_file_txn_from_json(struct ovsdb *,
+                                                    const struct json *,
+                                                    struct ovsdb_txn **);
+static void ovsdb_file_replica_create(struct ovsdb *, struct ovsdb_log *);
+
+struct ovsdb_error *
+ovsdb_file_open(const char *file_name, bool read_only, struct ovsdb **dbp)
+{
+    struct ovsdb_schema *schema;
+    struct ovsdb_error *error;
+    struct ovsdb_log *log;
+    struct json *json;
+    struct ovsdb *db;
+
+    error = ovsdb_log_open(file_name, read_only ? O_RDONLY : O_RDWR, &log);
+    if (error) {
+        return error;
+    }
+
+    error = ovsdb_log_read(log, &json);
+    if (error) {
+        return error;
+    } else if (!json) {
+        return ovsdb_io_error(EOF, "%s: database file contains no schema",
+                              file_name);
+    }
+
+    error = ovsdb_schema_from_json(json, &schema);
+    if (error) {
+        json_destroy(json);
+        return ovsdb_wrap_error(error,
+                                "failed to parse \"%s\" as ovsdb schema",
+                                file_name);
+    }
+    json_destroy(json);
+
+    db = ovsdb_create(schema);
+    while ((error = ovsdb_log_read(log, &json)) == NULL && json) {
+        struct ovsdb_txn *txn;
+
+        error = ovsdb_file_txn_from_json(db, json, &txn);
+        json_destroy(json);
+        if (error) {
+            break;
+        }
+
+        ovsdb_txn_commit(txn, false);
+    }
+    if (error) {
+        char *msg = ovsdb_error_to_string(error);
+        VLOG_WARN("%s", msg);
+        free(msg);
+
+        ovsdb_error_destroy(error);
+    }
+
+    if (!read_only) {
+        ovsdb_file_replica_create(db, log);
+    } else {
+        ovsdb_log_close(log);
+    }
+
+    *dbp = db;
+    return NULL;
+}
+
+static struct ovsdb_error *
+ovsdb_file_txn_row_from_json(struct ovsdb_txn *txn, struct ovsdb_table *table,
+                             const struct uuid *row_uuid, struct json *json)
+{
+    const struct ovsdb_row *row = ovsdb_table_get_row(table, row_uuid);
+    if (json->type == JSON_NULL) {
+        if (!row) {
+            return ovsdb_syntax_error(NULL, NULL, "transaction deletes "
+                                      "row "UUID_FMT" that does not exist",
+                                      UUID_ARGS(row_uuid));
+        }
+        ovsdb_txn_row_delete(txn, row);
+        return NULL;
+    } else if (row) {
+        return ovsdb_row_from_json(ovsdb_txn_row_modify(txn, row),
+                                   json, NULL, NULL);
+    } else {
+        struct ovsdb_error *error;
+        struct ovsdb_row *new;
+
+        new = ovsdb_row_create(table);
+        *ovsdb_row_get_uuid_rw(new) = *row_uuid;
+        error = ovsdb_row_from_json(new, json, NULL, NULL);
+        if (error) {
+            ovsdb_row_destroy(new);
+        }
+
+        ovsdb_txn_row_insert(txn, new);
+
+        return error;
+    }
+}
+
+static struct ovsdb_error *
+ovsdb_file_txn_table_from_json(struct ovsdb_txn *txn,
+                               struct ovsdb_table *table, struct json *json)
+{
+    struct shash_node *node;
+
+    if (json->type != JSON_OBJECT) {
+        return ovsdb_syntax_error(json, NULL, "object expected");
+    }
+
+    SHASH_FOR_EACH (node, json->u.object) {
+        const char *uuid_string = node->name;
+        struct json *txn_row_json = node->data;
+        struct ovsdb_error *error;
+        struct uuid row_uuid;
+
+        if (!uuid_from_string(&row_uuid, uuid_string)) {
+            return ovsdb_syntax_error(json, NULL, "\"%s\" is not a valid UUID",
+                                      uuid_string);
+        }
+
+        error = ovsdb_file_txn_row_from_json(txn, table, &row_uuid,
+                                             txn_row_json);
+        if (error) {
+            return error;
+        }
+    }
+
+    return NULL;
+}
+
+static struct ovsdb_error *
+ovsdb_file_txn_from_json(struct ovsdb *db, const struct json *json,
+                         struct ovsdb_txn **txnp)
+{
+    struct ovsdb_error *error;
+    struct shash_node *node;
+    struct ovsdb_txn *txn;
+
+    *txnp = NULL;
+    if (json->type != JSON_OBJECT) {
+        return ovsdb_syntax_error(json, NULL, "object expected");
+    }
+
+    txn = ovsdb_txn_create(db);
+    SHASH_FOR_EACH (node, json->u.object) {
+        const char *table_name = node->name;
+        struct json *txn_table_json = node->data;
+        struct ovsdb_table *table;
+
+        table = shash_find_data(&db->tables, table_name);
+        if (!table) {
+            error = ovsdb_syntax_error(json, "unknown table",
+                                       "No table named %s.", table_name);
+            goto error;
+        }
+
+        error = ovsdb_file_txn_table_from_json(txn, table, txn_table_json);
+        if (error) {
+            goto error;
+        }
+    }
+    *txnp = txn;
+    return NULL;
+
+error:
+    ovsdb_txn_abort(txn);
+    return error;
+}
+
+/* Replica implementation. */
+
+struct ovsdb_file_replica {
+    struct ovsdb_replica replica;
+    struct ovsdb_log *log;
+};
+
+static const struct ovsdb_replica_class ovsdb_file_replica_class;
+
+static void
+ovsdb_file_replica_create(struct ovsdb *db, struct ovsdb_log *log)
+{
+    struct ovsdb_file_replica *r = xmalloc(sizeof *r);
+    ovsdb_replica_init(&r->replica, &ovsdb_file_replica_class);
+    r->log = log;
+    ovsdb_add_replica(db, &r->replica);
+
+}
+
+static struct ovsdb_file_replica *
+ovsdb_file_replica_cast(struct ovsdb_replica *replica)
+{
+    assert(replica->class == &ovsdb_file_replica_class);
+    return CONTAINER_OF(replica, struct ovsdb_file_replica, replica);
+}
+
+struct ovsdb_file_replica_aux {
+    struct json *json;          /* JSON for the whole transaction. */
+    struct json *table_json;    /* JSON for 'table''s transaction. */
+    struct ovsdb_table *table;  /* Table described in 'table_json'.  */
+};
+
+static bool
+ovsdb_file_replica_change_cb(const struct ovsdb_row *old,
+                             const struct ovsdb_row *new,
+                             void *aux_)
+{
+    struct ovsdb_file_replica_aux *aux = aux_;
+    struct json *row;
+
+    if (!new) {
+        row = json_null_create();
+    } else {
+        struct shash_node *node;
+
+        row = NULL;
+        SHASH_FOR_EACH (node, &new->table->schema->columns) {
+            const struct ovsdb_column *column = node->data;
+            const struct ovsdb_type *type = &column->type;
+            unsigned int idx = column->index;
+
+            if (idx != OVSDB_COL_UUID && column->persistent
+                && (!old || !ovsdb_datum_equals(&old->fields[idx],
+                                                &new->fields[idx], type)))
+            {
+                if (!row) {
+                    row = json_object_create();
+                }
+                json_object_put(row, column->name,
+                                ovsdb_datum_to_json(&new->fields[idx], type));
+            }
+        }
+    }
+
+    if (row) {
+        struct ovsdb_table *table = new ? new->table : old->table;
+        char uuid[UUID_LEN + 1];
+
+        if (table != aux->table) {
+            /* Create JSON object for transaction overall. */
+            if (!aux->json) {
+                aux->json = json_object_create();
+            }
+
+            /* Create JSON object for transaction on this table. */
+            aux->table_json = json_object_create();
+            aux->table = table;
+            json_object_put(aux->json, table->schema->name, aux->table_json);
+        }
+
+        /* Add row to transaction for this table. */
+        snprintf(uuid, sizeof uuid,
+                 UUID_FMT, UUID_ARGS(ovsdb_row_get_uuid(new ? new : old)));
+        json_object_put(aux->table_json, uuid, row);
+    }
+
+    return true;
+}
+
+static struct ovsdb_error *
+ovsdb_file_replica_commit(struct ovsdb_replica *r_,
+                          const struct ovsdb_txn *txn, bool durable)
+{
+    struct ovsdb_file_replica *r = ovsdb_file_replica_cast(r_);
+    struct ovsdb_file_replica_aux aux;
+    struct ovsdb_error *error;
+
+    aux.json = NULL;
+    aux.table_json = NULL;
+    aux.table = NULL;
+    ovsdb_txn_for_each_change(txn, ovsdb_file_replica_change_cb, &aux);
+
+    if (!aux.json) {
+        /* Nothing to commit. */
+        return NULL;
+    }
+
+    error = ovsdb_log_write(r->log, aux.json);
+    json_destroy(aux.json);
+    if (error) {
+        return ovsdb_wrap_error(error, "writing transaction failed");
+    }
+
+    if (durable) {
+        error = ovsdb_log_commit(r->log);
+        if (error) {
+            return ovsdb_wrap_error(error, "committing transaction failed");
+        }
+    }
+
+    return NULL;
+}
+
+static void
+ovsdb_file_replica_destroy(struct ovsdb_replica *r_)
+{
+    struct ovsdb_file_replica *r = ovsdb_file_replica_cast(r_);
+
+    ovsdb_log_close(r->log);
+    free(r);
+}
+
+static const struct ovsdb_replica_class ovsdb_file_replica_class = {
+    ovsdb_file_replica_commit,
+    ovsdb_file_replica_destroy
+};
diff --git a/ovsdb/file.h b/ovsdb/file.h
new file mode 100644
index 000000000..2a2747798
--- /dev/null
+++ b/ovsdb/file.h
@@ -0,0 +1,28 @@
+/* Copyright (c) 2009 Nicira Networks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef OVSDB_FILE_H
+#define OVSDB_FILE_H 1
+
+#include <stdbool.h>
+#include "compiler.h"
+
+struct ovsdb;
+
+struct ovsdb_error *ovsdb_file_open(const char *file_name, bool read_only,
+                                    struct ovsdb **)
+    WARN_UNUSED_RESULT;
+
+#endif /* ovsdb/file.h */
diff --git a/ovsdb/ovsdb-server.c b/ovsdb/ovsdb-server.c
index 908042d77..a91e77893 100644
--- a/ovsdb/ovsdb-server.c
+++ b/ovsdb/ovsdb-server.c
@@ -24,6 +24,7 @@
 #include "command-line.h"
 #include "daemon.h"
 #include "fault.h"
+#include "file.h"
 #include "json.h"
 #include "jsonrpc.h"
 #include "jsonrpc-server.h"
@@ -68,7 +69,7 @@ main(int argc, char *argv[])
 
     parse_options(argc, argv, &file_name, &active, &passive);
 
-    error = ovsdb_open(file_name, false, &db);
+    error = ovsdb_file_open(file_name, false, &db);
     if (error) {
         ovs_fatal(0, "%s", ovsdb_error_to_string(error));
     }
diff --git a/ovsdb/ovsdb-tool.c b/ovsdb/ovsdb-tool.c
index 8b81ef615..61aea75dc 100644
--- a/ovsdb/ovsdb-tool.c
+++ b/ovsdb/ovsdb-tool.c
@@ -24,6 +24,7 @@
 
 #include "command-line.h"
 #include "compiler.h"
+#include "file.h"
 #include "log.h"
 #include "json.h"
 #include "ovsdb.h"
@@ -167,7 +168,7 @@ transact(bool read_only, const char *db_file_name, const char *transaction)
     struct json *request, *result;
     struct ovsdb *db;
 
-    check_ovsdb_error(ovsdb_open(db_file_name, read_only, &db));
+    check_ovsdb_error(ovsdb_file_open(db_file_name, read_only, &db));
 
     request = parse_json(transaction);
     result = ovsdb_execute(db, request, 0, NULL);
diff --git a/ovsdb/ovsdb.c b/ovsdb/ovsdb.c
index 7caa229da..4d5f1c5e5 100644
--- a/ovsdb/ovsdb.c
+++ b/ovsdb/ovsdb.c
@@ -17,18 +17,12 @@
 
 #include "ovsdb.h"
 
-#include <fcntl.h>
-
-#include "log.h"
 #include "json.h"
 #include "ovsdb-error.h"
 #include "ovsdb-parser.h"
 #include "table.h"
 #include "transaction.h"
 
-#define THIS_MODULE VLM_ovsdb
-#include "vlog.h"
-
 struct ovsdb_schema *
 ovsdb_schema_create(const char *name, const char *comment)
 {
@@ -153,14 +147,14 @@ ovsdb_schema_to_json(const struct ovsdb_schema *schema)
 }
 
 struct ovsdb *
-ovsdb_create(struct ovsdb_log *log, struct ovsdb_schema *schema)
+ovsdb_create(struct ovsdb_schema *schema)
 {
     struct shash_node *node;
     struct ovsdb *db;
 
     db = xmalloc(sizeof *db);
     db->schema = schema;
-    db->log = log;
+    list_init(&db->replicas);
     list_init(&db->triggers);
     db->run_triggers = false;
 
@@ -173,71 +167,20 @@ ovsdb_create(struct ovsdb_log *log, struct ovsdb_schema *schema)
     return db;
 }
 
-struct ovsdb_error *
-ovsdb_open(const char *file_name, bool read_only, struct ovsdb **dbp)
-{
-    struct ovsdb_schema *schema;
-    struct ovsdb_error *error;
-    struct ovsdb_log *log;
-    struct json *json;
-    struct ovsdb *db;
-
-    error = ovsdb_log_open(file_name, read_only ? O_RDONLY : O_RDWR, &log);
-    if (error) {
-        return error;
-    }
-
-    error = ovsdb_log_read(log, &json);
-    if (error) {
-        return error;
-    } else if (!json) {
-        return ovsdb_io_error(EOF, "%s: database file contains no schema",
-                              file_name);
-    }
-
-    error = ovsdb_schema_from_json(json, &schema);
-    if (error) {
-        json_destroy(json);
-        return ovsdb_wrap_error(error,
-                                "failed to parse \"%s\" as ovsdb schema",
-                                file_name);
-    }
-    json_destroy(json);
-
-    db = ovsdb_create(read_only ? NULL : log, schema);
-    while ((error = ovsdb_log_read(log, &json)) == NULL && json) {
-        struct ovsdb_txn *txn;
-
-        error = ovsdb_txn_from_json(db, json, &txn);
-        json_destroy(json);
-        if (error) {
-            break;
-        }
-
-        ovsdb_txn_commit(txn);
-    }
-    if (error) {
-        char *msg = ovsdb_error_to_string(error);
-        VLOG_WARN("%s", msg);
-        free(msg);
-
-        ovsdb_error_destroy(error);
-    }
-
-    if (read_only) {
-        ovsdb_log_close(log);
-    }
-
-    *dbp = db;
-    return NULL;
-}
-
 void
 ovsdb_destroy(struct ovsdb *db)
 {
     if (db) {
         struct shash_node *node;
 
+        /* Remove all the replicas. */
+        while (!list_is_empty(&db->replicas)) {
+            struct ovsdb_replica *r
+                = CONTAINER_OF(list_pop_back(&db->replicas),
+                               struct ovsdb_replica, node);
+            ovsdb_remove_replica(db, r);
+        }
+
         /* Delete all the tables.  This also deletes their schemas. */
         SHASH_FOR_EACH (node, &db->tables) {
             struct ovsdb_table *table = node->data;
@@ -251,7 +194,6 @@ ovsdb_destroy(struct ovsdb *db)
         shash_clear(&db->schema->tables);
 
         ovsdb_schema_destroy(db->schema);
-        ovsdb_log_close(db->log);
         free(db);
     }
 }
@@ -261,3 +203,23 @@ ovsdb_get_table(const struct ovsdb *db, const char *name)
 {
     return shash_find_data(&db->tables, name);
 }
+
+void
+ovsdb_replica_init(struct ovsdb_replica *r,
+                   const struct ovsdb_replica_class *class)
+{
+    r->class = class;
+}
+
+void
+ovsdb_add_replica(struct ovsdb *db, struct ovsdb_replica *r)
+{
+    list_push_back(&db->replicas, &r->node);
+}
+
+void
+ovsdb_remove_replica(struct ovsdb *db UNUSED, struct ovsdb_replica *r)
+{
+    list_remove(&r->node);
+    (r->class->destroy)(r);
+}
diff --git a/ovsdb/ovsdb.h b/ovsdb/ovsdb.h
index d57ebfca6..24ebd9c67 100644
--- a/ovsdb/ovsdb.h
+++ b/ovsdb/ovsdb.h
@@ -22,6 +22,8 @@
 #include "shash.h"
 
 struct json;
+struct ovsdb_log;
+struct ovsdb_txn;
 struct uuid;
 
 /* Database schema. */
@@ -46,7 +48,7 @@ struct json *ovsdb_schema_to_json(const struct ovsdb_schema *);
 /* Database. */
 struct ovsdb {
     struct ovsdb_schema *schema;
-    struct ovsdb_log *log;      /* Disk log (null for in-memory db). */
+    struct list replicas;       /* Contains "struct ovsdb_replica"s. */
     struct shash tables;        /* Contains "struct ovsdb_table *"s. */
 
     /* Triggers. */
@@ -54,10 +56,7 @@ struct ovsdb {
     bool run_triggers;
 };
 
-struct ovsdb *ovsdb_create(struct ovsdb_log *, struct ovsdb_schema *);
-struct ovsdb_error *ovsdb_open(const char *file_name, bool read_only,
-                               struct ovsdb **)
-    WARN_UNUSED_RESULT;
+struct ovsdb *ovsdb_create(struct ovsdb_schema *);
 void ovsdb_destroy(struct ovsdb *);
 
 struct ovsdb_error *ovsdb_from_json(const struct json *, struct ovsdb **)
@@ -69,5 +68,24 @@ struct ovsdb_table *ovsdb_get_table(const struct ovsdb *, const char *);
 struct json *ovsdb_execute(struct ovsdb *, const struct json *params,
                            long long int elapsed_msec,
                            long long int *timeout_msec);
+
+/* Database replication. */
+
+struct ovsdb_replica {
+    struct list node;           /* Element in "struct ovsdb" replicas list. */
+    const struct ovsdb_replica_class *class;
+};
+
+struct ovsdb_replica_class {
+    struct ovsdb_error *(*commit)(struct ovsdb_replica *,
+                                  const struct ovsdb_txn *, bool durable);
+    void (*destroy)(struct ovsdb_replica *);
+};
+
+void ovsdb_replica_init(struct ovsdb_replica *,
+                        const struct ovsdb_replica_class *);
+
+void ovsdb_add_replica(struct ovsdb *, struct ovsdb_replica *);
+void ovsdb_remove_replica(struct ovsdb *, struct ovsdb_replica *);
 
 #endif /* ovsdb/ovsdb.h */
diff --git a/ovsdb/transaction.c b/ovsdb/transaction.c
index 21a46ec7c..d5e360167 100644
--- a/ovsdb/transaction.c
+++ b/ovsdb/transaction.c
@@ -59,13 +59,6 @@ struct ovsdb_txn_row {
     struct ovsdb_row *new;      /* The new row. */
 };
 
-static const struct uuid *
-ovsdb_txn_row_get_uuid(const struct ovsdb_txn_row *txn_row)
-{
-    const struct ovsdb_row *row = txn_row->old ? txn_row->old : txn_row->new;
-    return ovsdb_row_get_uuid(row);
-}
-
 struct ovsdb_txn *
 ovsdb_txn_create(struct ovsdb *db)
 {
@@ -131,197 +124,43 @@ ovsdb_txn_row_commit(struct ovsdb_txn_row *txn_row)
     ovsdb_row_destroy(txn_row->old);
 }
 
-void
-ovsdb_txn_commit(struct ovsdb_txn *txn)
-{
-    txn->db->run_triggers = true;
-    ovsdb_txn_destroy(txn, ovsdb_txn_row_commit);
-}
-
-static void
-put_json_column(struct json *object, const struct ovsdb_row *row,
-                const struct ovsdb_column *column)
-{
-    json_object_put(object, column->name,
-                    ovsdb_datum_to_json(&row->fields[column->index],
-                                        &column->type));
-}
-
-static struct json *
-ovsdb_txn_row_to_json(const struct ovsdb_txn_row *txn_row)
-{
-    const struct ovsdb_row *old = txn_row->old;
-    const struct ovsdb_row *new = txn_row->new;
-    struct shash_node *node;
-    struct json *json;
-
-    if (!new) {
-        return json_null_create();
-    }
-
-    json = NULL;
-    SHASH_FOR_EACH (node, &new->table->schema->columns) {
-        struct ovsdb_column *column = node->data;
-        unsigned int index = column->index;
-
-        if (index != OVSDB_COL_UUID && column->persistent
-            && (!old || !ovsdb_datum_equals(&old->fields[index],
-                                            &new->fields[index],
-                                            &column->type)))
-        {
-            if (!json) {
-                json = json_object_create();
-            }
-            put_json_column(json, new, column);
-        }
-    }
-    return json;
-}
-
-static struct json *
-ovsdb_txn_table_to_json(const struct ovsdb_txn_table *txn_table)
-{
-    struct ovsdb_txn_row *txn_row;
-    struct json *txn_table_json;
-
-    txn_table_json = NULL;
-    HMAP_FOR_EACH (txn_row, struct ovsdb_txn_row, hmap_node,
-                   &txn_table->txn_rows) {
-        struct json *txn_row_json = ovsdb_txn_row_to_json(txn_row);
-        if (txn_row_json) {
-            char uuid[UUID_LEN + 1];
-
-            if (!txn_table_json) {
-                txn_table_json = json_object_create();
-            }
-
-            snprintf(uuid, sizeof uuid,
-                     UUID_FMT, UUID_ARGS(ovsdb_txn_row_get_uuid(txn_row)));
-            json_object_put(txn_table_json, uuid, txn_row_json);
-        }
-    }
-    return txn_table_json;
-}
-
-struct json *
-ovsdb_txn_to_json(const struct ovsdb_txn *txn)
-{
-    struct ovsdb_txn_table *txn_table;
-    struct json *txn_json;
-
-    txn_json = NULL;
-    HMAP_FOR_EACH (txn_table, struct ovsdb_txn_table, hmap_node,
-                   &txn->txn_tables) {
-        struct json *txn_table_json = ovsdb_txn_table_to_json(txn_table);
-        if (!txn_json) {
-            txn_json = json_object_create();
-        }
-        json_object_put(txn_json, txn_table->table->schema->name,
-                        txn_table_json);
-    }
-    return txn_json;
-}
-
-static struct ovsdb_error *
-ovsdb_txn_row_from_json(struct ovsdb_txn *txn, struct ovsdb_table *table,
-                        const struct uuid *row_uuid, struct json *json)
+struct ovsdb_error *
+ovsdb_txn_commit(struct ovsdb_txn *txn, bool durable)
 {
-    const struct ovsdb_row *row = ovsdb_table_get_row(table, row_uuid);
-    if (json->type == JSON_NULL) {
-        if (!row) {
-            return ovsdb_syntax_error(NULL, NULL, "transaction deletes "
-                                      "row "UUID_FMT" that does not exist",
-                                      UUID_ARGS(row_uuid));
-        }
-        ovsdb_txn_row_delete(txn, row);
-        return NULL;
-    } else if (row) {
-        return ovsdb_row_from_json(ovsdb_txn_row_modify(txn, row),
-                                   json, NULL, NULL);
-    } else {
-        struct ovsdb_error *error;
-        struct ovsdb_row *new;
+    struct ovsdb_replica *replica;
+    struct ovsdb_error *error;
 
-        new = ovsdb_row_create(table);
-        *ovsdb_row_get_uuid_rw(new) = *row_uuid;
-        error = ovsdb_row_from_json(new, json, NULL, NULL);
+    LIST_FOR_EACH (replica, struct ovsdb_replica, node, &txn->db->replicas) {
+        error = (replica->class->commit)(replica, txn, durable);
         if (error) {
-            ovsdb_row_destroy(new);
-        }
-
-        ovsdb_txn_row_insert(txn, new);
-
-        return error;
-    }
-}
-
-static struct ovsdb_error *
-ovsdb_txn_table_from_json(struct ovsdb_txn *txn, struct ovsdb_table *table,
-                          struct json *json)
-{
-    struct shash_node *node;
-
-    if (json->type != JSON_OBJECT) {
-        return ovsdb_syntax_error(json, NULL, "object expected");
-    }
-
-    SHASH_FOR_EACH (node, json->u.object) {
-        const char *uuid_string = node->name;
-        struct json *txn_row_json = node->data;
-        struct ovsdb_error *error;
-        struct uuid row_uuid;
+            /* We don't support two-phase commit so only the first replica is
+             * allowed to report an error. */
+            assert(&replica->node == txn->db->replicas.next);
 
-        if (!uuid_from_string(&row_uuid, uuid_string)) {
-            return ovsdb_syntax_error(json, NULL, "\"%s\" is not a valid UUID",
-                                      uuid_string);
-        }
-
-        error = ovsdb_txn_row_from_json(txn, table, &row_uuid, txn_row_json);
-        if (error) {
+            ovsdb_txn_abort(txn);
             return error;
         }
     }
 
+    txn->db->run_triggers = true;
+    ovsdb_txn_destroy(txn, ovsdb_txn_row_commit);
     return NULL;
 }
 
-struct ovsdb_error *
-ovsdb_txn_from_json(struct ovsdb *db, const struct json *json,
-                    struct ovsdb_txn **txnp)
+void
+ovsdb_txn_for_each_change(const struct ovsdb_txn *txn,
+                          ovsdb_txn_row_cb_func *cb, void *aux)
 {
-    struct ovsdb_error *error;
-    struct shash_node *node;
-    struct ovsdb_txn *txn;
+    struct ovsdb_txn_table *t;
+    struct ovsdb_txn_row *r;
 
-    *txnp = NULL;
-    if (json->type != JSON_OBJECT) {
-        return ovsdb_syntax_error(json, NULL, "object expected");
-    }
-
-    txn = ovsdb_txn_create(db);
-    SHASH_FOR_EACH (node, json->u.object) {
-        const char *table_name = node->name;
-        struct json *txn_table_json = node->data;
-        struct ovsdb_table *table;
-
-        table = shash_find_data(&db->tables, table_name);
-        if (!table) {
-            error = ovsdb_syntax_error(json, "unknown table",
-                                       "No table named %s.", table_name);
-            goto error;
-        }
-
-        error = ovsdb_txn_table_from_json(txn, table, txn_table_json);
-        if (error) {
-            goto error;
+    HMAP_FOR_EACH (t, struct ovsdb_txn_table, hmap_node, &txn->txn_tables) {
+        HMAP_FOR_EACH (r, struct ovsdb_txn_row, hmap_node, &t->txn_rows) {
+            if (!cb(r->old, r->new, aux)) {
+                break;
+            }
         }
-    }
-    *txnp = txn;
-    return NULL;
-
-error:
-    ovsdb_txn_abort(txn);
-    return error;
+   }
 }
 
 static struct ovsdb_txn_table *
diff --git a/ovsdb/transaction.h b/ovsdb/transaction.h
index 293eaf4ff..048bf74fa 100644
--- a/ovsdb/transaction.h
+++ b/ovsdb/transaction.h
@@ -19,23 +19,24 @@
 #include <stdbool.h>
 #include "compiler.h"
 
+struct json;
 struct ovsdb;
 struct ovsdb_table;
 struct uuid;
 
 struct ovsdb_txn *ovsdb_txn_create(struct ovsdb *);
 void ovsdb_txn_abort(struct ovsdb_txn *);
-void ovsdb_txn_commit(struct ovsdb_txn *);
-
-struct json *ovsdb_txn_to_json(const struct ovsdb_txn *);
-struct ovsdb_error *ovsdb_txn_from_json(struct ovsdb *, const struct json *,
-                                        struct ovsdb_txn **)
-    WARN_UNUSED_RESULT;
+struct ovsdb_error *ovsdb_txn_commit(struct ovsdb_txn *, bool durable);
 
 struct ovsdb_row *ovsdb_txn_row_modify(struct ovsdb_txn *,
                                        const struct ovsdb_row *);
-
 void ovsdb_txn_row_insert(struct ovsdb_txn *, struct ovsdb_row *);
 void ovsdb_txn_row_delete(struct ovsdb_txn *, const struct ovsdb_row *);
 
+typedef bool ovsdb_txn_row_cb_func(const struct ovsdb_row *old,
+                                   const struct ovsdb_row *new,
+                                   void *aux);
+void ovsdb_txn_for_each_change(const struct ovsdb_txn *,
+                               ovsdb_txn_row_cb_func *, void *aux);
+
 #endif /* ovsdb/transaction.h */