X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fovsdb-idl.c;h=3f3ce555ac3f4ec413352107ef65e40730ade696;hb=3fe8053b36da715c411b907ac125e41f9e9a28f2;hp=7ba0fd68366b19969a220f0edde6302b9080e970;hpb=c7f7adb70cb34bb568f9b6327d6adb945326e029;p=sliver-openvswitch.git

diff --git a/lib/ovsdb-idl.c b/lib/ovsdb-idl.c
index 7ba0fd683..3f3ce555a 100644
--- a/lib/ovsdb-idl.c
+++ b/lib/ovsdb-idl.c
@@ -25,6 +25,7 @@
 
 #include "bitmap.h"
 #include "dynamic-string.h"
+#include "fatal-signal.h"
 #include "json.h"
 #include "jsonrpc.h"
 #include "ovsdb-data.h"
@@ -81,6 +82,7 @@ struct ovsdb_idl_txn {
     struct ovsdb_idl *idl;
     struct hmap txn_rows;
     enum ovsdb_idl_txn_status status;
+    char *error;
     bool dry_run;
     struct ds comment;
 
@@ -90,6 +92,16 @@ struct ovsdb_idl_txn {
     struct json *inc_where;
     unsigned int inc_index;
     int64_t inc_new_value;
+
+    /* Inserted rows. */
+    struct hmap inserted_rows;
+};
+
+struct ovsdb_idl_txn_insert {
+    struct hmap_node hmap_node; /* In struct ovsdb_idl_txn's inserted_rows. */
+    struct uuid dummy;          /* Dummy UUID used locally. */
+    int op_index;               /* Index into transaction's operation array. */
+    struct uuid real;           /* Real UUID used by database server. */
 };
 
 static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);
@@ -115,6 +127,8 @@ static struct ovsdb_idl_row *ovsdb_idl_row_create(struct ovsdb_idl_table *,
                                                   const struct uuid *);
 static void ovsdb_idl_row_destroy(struct ovsdb_idl_row *);
 
+static void ovsdb_idl_row_parse(struct ovsdb_idl_row *);
+static void ovsdb_idl_row_unparse(struct ovsdb_idl_row *);
 static void ovsdb_idl_row_clear_old(struct ovsdb_idl_row *);
 static void ovsdb_idl_row_clear_new(struct ovsdb_idl_row *);
 
@@ -199,10 +213,8 @@ ovsdb_idl_clear(struct ovsdb_idl *idl)
             struct ovsdb_idl_arc *arc, *next_arc;
 
             if (!ovsdb_idl_row_is_orphan(row)) {
-                (row->table->class->unparse)(row);
-                ovsdb_idl_row_clear_old(row);
+                ovsdb_idl_row_unparse(row);
             }
-            hmap_remove(&table->rows, &row->hmap_node);
             LIST_FOR_EACH_SAFE (arc, next_arc, struct ovsdb_idl_arc, src_node,
                                 &row->src_arcs) {
                 free(arc);
@@ -210,7 +222,7 @@ ovsdb_idl_clear(struct ovsdb_idl *idl)
             /* No need to do anything with dst_arcs: some node has those arcs
              * as forward arcs and will destroy them itself. */
 
-            free(row);
+            ovsdb_idl_row_destroy(row);
         }
     }
 
@@ -331,7 +343,9 @@ ovsdb_idl_send_monitor_request(struct ovsdb_idl *idl)
 
     json_destroy(idl->monitor_request_id);
     msg = jsonrpc_create_request(
-        "monitor", json_array_create_2(json_null_create(), monitor_requests),
+        "monitor",
+        json_array_create_3(json_string_create(idl->class->database),
+                            json_null_create(), monitor_requests),
         &idl->monitor_request_id);
     jsonrpc_session_send(idl->session, msg);
 }
@@ -528,10 +542,65 @@ ovsdb_idl_row_update(struct ovsdb_idl_row *row, const struct json *row_json)
     }
 }
 
+/* When a row A refers to row B through a column with a "refTable" constraint,
+ * but row B does not exist, row B is called an "orphan row".  Orphan rows
+ * should not persist, because the database enforces referential integrity, but
+ * they can appear transiently as changes from the database are received (the
+ * database doesn't try to topologically sort them and circular references mean
+ * it isn't always possible anyhow).
+ *
+ * This function returns true if 'row' is an orphan row, otherwise false.
+ */
 static bool
 ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *row)
 {
-    return !row->old;
+    return !row->old && !row->new;
+}
+
+/* Returns true if 'row' is conceptually part of the database as modified by
+ * the current transaction (if any), false otherwise.
+ *
+ * This function will return true if 'row' is not an orphan (see the comment on
+ * ovsdb_idl_row_is_orphan()) and:
+ *
+ *   - 'row' exists in the database and has not been deleted within the
+ *     current transaction (if any).
+ *
+ *   - 'row' was inserted within the current transaction and has not been
+ *     deleted.  (In the latter case you should not have passed 'row' in at
+ *     all, because ovsdb_idl_txn_delete() freed it.)
+ *
+ * This function will return false if 'row' is an orphan or if 'row' was
+ * deleted within the current transaction.
+ */
+static bool
+ovsdb_idl_row_exists(const struct ovsdb_idl_row *row)
+{
+    return row->new != NULL;
+}
+
+static void
+ovsdb_idl_row_parse(struct ovsdb_idl_row *row)
+{
+    const struct ovsdb_idl_table_class *class = row->table->class;
+    size_t i;
+
+    for (i = 0; i < class->n_columns; i++) {
+        const struct ovsdb_idl_column *c = &class->columns[i];
+        (c->parse)(row, &row->old[i]);
+    }
+}
+
+static void
+ovsdb_idl_row_unparse(struct ovsdb_idl_row *row)
+{
+    const struct ovsdb_idl_table_class *class = row->table->class;
+    size_t i;
+
+    for (i = 0; i < class->n_columns; i++) {
+        const struct ovsdb_idl_column *c = &class->columns[i];
+        (c->unparse)(row);
+    }
 }
 
 static void
@@ -591,15 +660,16 @@ ovsdb_idl_row_clear_arcs(struct ovsdb_idl_row *row, bool destroy_dsts)
 
 /* Force nodes that reference 'row' to reparse. */
 static void
-ovsdb_idl_row_reparse_backrefs(struct ovsdb_idl_row *row, bool destroy_dsts)
+ovsdb_idl_row_reparse_backrefs(struct ovsdb_idl_row *row)
 {
     struct ovsdb_idl_arc *arc, *next;
 
     /* This is trickier than it looks.  ovsdb_idl_row_clear_arcs() will destroy
      * 'arc', so we need to use the "safe" variant of list traversal.  However,
-     * calling ref->table->class->parse will add an arc equivalent to 'arc' to
-     * row->arcs.  That could be a problem for traversal, but it adds it at the
-     * beginning of the list to prevent us from stumbling upon it again.
+     * calling an ovsdb_idl_column's 'parse' function will add an arc
+     * equivalent to 'arc' to row->arcs.  That could be a problem for
+     * traversal, but it adds it at the beginning of the list to prevent us
+     * from stumbling upon it again.
      *
      * (If duplicate arcs were possible then we would need to make sure that
      * 'next' didn't also point into 'arc''s destination, but we forbid
@@ -608,9 +678,9 @@ ovsdb_idl_row_reparse_backrefs(struct ovsdb_idl_row *row, bool destroy_dsts)
                         &row->dst_arcs) {
         struct ovsdb_idl_row *ref = arc->src;
 
-        (ref->table->class->unparse)(ref);
-        ovsdb_idl_row_clear_arcs(ref, destroy_dsts);
-        (ref->table->class->parse)(ref);
+        ovsdb_idl_row_unparse(ref);
+        ovsdb_idl_row_clear_arcs(ref, false);
+        ovsdb_idl_row_parse(ref);
     }
 }
 
@@ -618,7 +688,6 @@ static struct ovsdb_idl_row *
 ovsdb_idl_row_create__(const struct ovsdb_idl_table_class *class)
 {
     struct ovsdb_idl_row *row = xzalloc(class->allocation_size);
-    memset(row, 0, sizeof *row);
     list_init(&row->src_arcs);
     list_init(&row->dst_arcs);
     hmap_node_nullify(&row->txn_node);
@@ -657,31 +726,31 @@ ovsdb_idl_insert_row(struct ovsdb_idl_row *row, const struct json *row_json)
         ovsdb_datum_init_default(&row->old[i], &class->columns[i].type);
     }
     ovsdb_idl_row_update(row, row_json);
-    (class->parse)(row);
+    ovsdb_idl_row_parse(row);
 
-    ovsdb_idl_row_reparse_backrefs(row, false);
+    ovsdb_idl_row_reparse_backrefs(row);
 }
 
 static void
 ovsdb_idl_delete_row(struct ovsdb_idl_row *row)
 {
-    (row->table->class->unparse)(row);
+    ovsdb_idl_row_unparse(row);
     ovsdb_idl_row_clear_arcs(row, true);
     ovsdb_idl_row_clear_old(row);
     if (list_is_empty(&row->dst_arcs)) {
         ovsdb_idl_row_destroy(row);
     } else {
-        ovsdb_idl_row_reparse_backrefs(row, true);
+        ovsdb_idl_row_reparse_backrefs(row);
     }
 }
 
 static void
 ovsdb_idl_modify_row(struct ovsdb_idl_row *row, const struct json *row_json)
 {
-    (row->table->class->unparse)(row);
+    ovsdb_idl_row_unparse(row);
     ovsdb_idl_row_clear_arcs(row, true);
     ovsdb_idl_row_update(row, row_json);
-    (row->table->class->parse)(row);
+    ovsdb_idl_row_parse(row);
 }
 
 static bool
@@ -698,7 +767,7 @@ may_add_arc(const struct ovsdb_idl_row *src, const struct ovsdb_idl_row *dst)
      *
      * We only need to test whether the first arc in dst->dst_arcs originates
      * at 'src', since we add all of the arcs from a given source in a clump
-     * (in a single call to a row's ->parse function) and new arcs are always
+     * (in a single call to ovsdb_idl_row_parse()) and new arcs are always
      * added at the front of the dst_arcs list. */
     if (list_is_empty(&dst->dst_arcs)) {
         return true;
@@ -726,22 +795,44 @@ ovsdb_idl_get_row_arc(struct ovsdb_idl_row *src,
 
     dst_table = ovsdb_idl_table_from_class(idl, dst_table_class);
     dst = ovsdb_idl_get_row(dst_table, dst_uuid);
-    if (!dst) {
-        dst = ovsdb_idl_row_create(dst_table, dst_uuid);
-    }
+    if (idl->txn) {
+        /* We're being called from ovsdb_idl_txn_write().  We must not update
+         * any arcs, because the transaction will be backed out at commit or
+         * abort time and we don't want our graph screwed up.
+         *
+         * Just return the destination row, if there is one and it has not been
+         * deleted. */
+        if (dst && (hmap_node_is_null(&dst->txn_node) || dst->new)) {
+            return dst;
+        }
+        return NULL;
+    } else {
+        /* We're being called from some other context.  Update the graph. */
+        if (!dst) {
+            dst = ovsdb_idl_row_create(dst_table, dst_uuid);
+        }
+
+        /* Add a new arc, if it wouldn't be a self-arc or a duplicate arc. */
+        if (may_add_arc(src, dst)) {
+            /* The arc *must* be added at the front of the dst_arcs list.  See
+             * ovsdb_idl_row_reparse_backrefs() for details. */
+            arc = xmalloc(sizeof *arc);
+            list_push_front(&src->src_arcs, &arc->src_node);
+            list_push_front(&dst->dst_arcs, &arc->dst_node);
+            arc->src = src;
+            arc->dst = dst;
+        }
 
-    /* Add a new arc, if it wouldn't be a self-arc or a duplicate arc. */
-    if (may_add_arc(src, dst)) {
-        /* The arc *must* be added at the front of the dst_arcs list.  See
-         * ovsdb_idl_row_reparse_backrefs() for details. */
-        arc = xmalloc(sizeof *arc);
-        list_push_front(&src->src_arcs, &arc->src_node);
-        list_push_front(&dst->dst_arcs, &arc->dst_node);
-        arc->src = src;
-        arc->dst = dst;
+        return !ovsdb_idl_row_is_orphan(dst) ? dst : NULL;
     }
+}
 
-    return !ovsdb_idl_row_is_orphan(dst) ? dst : NULL;
+const struct ovsdb_idl_row *
+ovsdb_idl_get_row_for_uuid(const struct ovsdb_idl *idl,
+                           const struct ovsdb_idl_table_class *tc,
+                           const struct uuid *uuid)
+{
+    return ovsdb_idl_get_row(ovsdb_idl_table_from_class(idl, tc), uuid);
 }
 
 static struct ovsdb_idl_row *
@@ -751,14 +842,14 @@ next_real_row(struct ovsdb_idl_table *table, struct hmap_node *node)
         struct ovsdb_idl_row *row;
 
         row = CONTAINER_OF(node, struct ovsdb_idl_row, hmap_node);
-        if (!ovsdb_idl_row_is_orphan(row)) {
+        if (ovsdb_idl_row_exists(row)) {
             return row;
         }
     }
     return NULL;
 }
 
-struct ovsdb_idl_row *
+const struct ovsdb_idl_row *
 ovsdb_idl_first_row(const struct ovsdb_idl *idl,
                     const struct ovsdb_idl_table_class *table_class)
 {
@@ -767,7 +858,7 @@ ovsdb_idl_first_row(const struct ovsdb_idl *idl,
     return next_real_row(table, hmap_first(&table->rows));
 }
 
-struct ovsdb_idl_row *
+const struct ovsdb_idl_row *
 ovsdb_idl_next_row(const struct ovsdb_idl_row *row)
 {
     struct ovsdb_idl_table *table = row->table;
@@ -807,24 +898,39 @@ ovsdb_idl_txn_create(struct ovsdb_idl *idl)
 
     assert(!idl->txn);
     idl->txn = txn = xmalloc(sizeof *txn);
+    txn->request_id = NULL;
     txn->idl = idl;
-    txn->status = TXN_INCOMPLETE;
     hmap_init(&txn->txn_rows);
+    txn->status = TXN_INCOMPLETE;
+    txn->error = NULL;
     txn->dry_run = false;
     ds_init(&txn->comment);
+
     txn->inc_table = NULL;
     txn->inc_column = NULL;
     txn->inc_where = NULL;
+
+    hmap_init(&txn->inserted_rows);
+
     return txn;
 }
 
+/* Appends 's', which is treated as a printf()-type format string, to the
+ * comments that will be passed to the OVSDB server when 'txn' is committed.
+ * (The comment will be committed to the OVSDB log, which "ovsdb-tool
+ * show-log" can print in a relatively human-readable form.) */
 void
-ovsdb_idl_txn_add_comment(struct ovsdb_idl_txn *txn, const char *s)
+ovsdb_idl_txn_add_comment(struct ovsdb_idl_txn *txn, const char *s, ...)
 {
+    va_list args;
+
     if (txn->comment.length) {
         ds_put_char(&txn->comment, '\n');
     }
-    ds_put_cstr(&txn->comment, s);
+
+    va_start(args, s);
+    ds_put_format_valist(&txn->comment, s, args);
+    va_end(args);
 }
 
 void
@@ -846,14 +952,23 @@ ovsdb_idl_txn_increment(struct ovsdb_idl_txn *txn, const char *table,
 void
 ovsdb_idl_txn_destroy(struct ovsdb_idl_txn *txn)
 {
+    struct ovsdb_idl_txn_insert *insert, *next;
+
+    json_destroy(txn->request_id);
     if (txn->status == TXN_INCOMPLETE) {
         hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
     }
     ovsdb_idl_txn_abort(txn);
     ds_destroy(&txn->comment);
+    free(txn->error);
     free(txn->inc_table);
     free(txn->inc_column);
     json_destroy(txn->inc_where);
+    HMAP_FOR_EACH_SAFE (insert, next, struct ovsdb_idl_txn_insert, hmap_node,
+                        &txn->inserted_rows) {
+        free(insert);
+    }
+    hmap_destroy(&txn->inserted_rows);
     free(txn);
 }
 
@@ -953,12 +1068,22 @@ ovsdb_idl_txn_disassemble(struct ovsdb_idl_txn *txn)
 {
     struct ovsdb_idl_row *row, *next;
 
+    /* This must happen early.  Otherwise, ovsdb_idl_row_parse() will call an
+     * ovsdb_idl_column's 'parse' function, which will call
+     * ovsdb_idl_get_row_arc(), which will seen that the IDL is in a
+     * transaction and fail to update the graph.  */
+    txn->idl->txn = NULL;
+
     HMAP_FOR_EACH_SAFE (row, next, struct ovsdb_idl_row, txn_node,
                         &txn->txn_rows) {
-        if (row->old && row->written) {
-            (row->table->class->unparse)(row);
-            ovsdb_idl_row_clear_arcs(row, false);
-            (row->table->class->parse)(row);
+        if (row->old) {
+            if (row->written) {
+                ovsdb_idl_row_unparse(row);
+                ovsdb_idl_row_clear_arcs(row, false);
+                ovsdb_idl_row_parse(row);
+            }
+        } else {
+            ovsdb_idl_row_unparse(row);
         }
         ovsdb_idl_row_clear_new(row);
 
@@ -970,6 +1095,10 @@ ovsdb_idl_txn_disassemble(struct ovsdb_idl_txn *txn)
 
         hmap_remove(&txn->txn_rows, &row->txn_node);
         hmap_node_nullify(&row->txn_node);
+        if (!row->old) {
+            hmap_remove(&row->table->rows, &row->hmap_node);
+            free(row);
+        }
     }
     hmap_destroy(&txn->txn_rows);
     hmap_init(&txn->txn_rows);
@@ -986,7 +1115,8 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
         return txn->status;
     }
 
-    operations = json_array_create_empty();
+    operations = json_array_create_1(
+        json_string_create(txn->idl->class->database));
 
     /* Add prerequisites and declarations of new rows. */
     HMAP_FOR_EACH (row, struct ovsdb_idl_row, txn_node, &txn->txn_rows) {
@@ -1017,16 +1147,6 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
                                                     &column->type));
             }
         }
-        if (row->new && !row->old) {
-            struct json *op;
-
-            op = json_object_create();
-            json_array_add(operations, op);
-            json_object_put_string(op, "op", "declare");
-            json_object_put(op, "uuid-name",
-                            json_string_create_nocopy(
-                                uuid_name_from_uuid(&row->uuid)));
-        }
     }
 
     /* Add updates. */
@@ -1054,9 +1174,18 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
             if (row->old) {
                 json_object_put(op, "where", where_uuid_equals(&row->uuid));
             } else {
+                struct ovsdb_idl_txn_insert *insert;
+
                 json_object_put(op, "uuid-name",
                                 json_string_create_nocopy(
                                     uuid_name_from_uuid(&row->uuid)));
+
+                insert = xmalloc(sizeof *insert);
+                insert->dummy = row->uuid;
+                insert->op_index = operations->u.array.n - 1;
+                uuid_zero(&insert->real);
+                hmap_insert(&txn->inserted_rows, &insert->hmap_node,
+                            uuid_hash(&insert->dummy));
             }
             row_json = json_object_create();
             json_object_put(op, "row", row_json);
@@ -1064,9 +1193,10 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
             BITMAP_FOR_EACH_1 (idx, class->n_columns, row->written) {
                 const struct ovsdb_idl_column *column = &class->columns[idx];
 
-                if (!row->old || !ovsdb_datum_equals(&row->old[idx],
-                                                     &row->new[idx],
-                                                     &column->type)) {
+                if (row->old
+                    ? !ovsdb_datum_equals(&row->old[idx], &row->new[idx],
+                                          &column->type)
+                    : !ovsdb_datum_is_default(&row->new[idx], &column->type)) {
                     json_object_put(row_json, column->name,
                                     substitute_uuids(
                                         ovsdb_datum_to_json(&row->new[idx],
@@ -1088,7 +1218,7 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
     if (txn->inc_table && any_updates) {
         struct json *op;
 
-        txn->inc_index = operations->u.array.n;
+        txn->inc_index = operations->u.array.n - 1;
 
         op = json_object_create();
         json_object_put_string(op, "op", "mutate");
@@ -1137,14 +1267,31 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
         hmap_insert(&txn->idl->outstanding_txns, &txn->hmap_node,
                     json_hash(txn->request_id, 0));
     } else {
-        txn->status = TXN_INCOMPLETE;
+        txn->status = TXN_TRY_AGAIN;
     }
 
-    txn->idl->txn = NULL;
     ovsdb_idl_txn_disassemble(txn);
     return txn->status;
 }
 
+/* Attempts to commit 'txn', blocking until the commit either succeeds or
+ * fails.  Returns the final commit status, which may be any TXN_* value other
+ * than TXN_INCOMPLETE. */
+enum ovsdb_idl_txn_status
+ovsdb_idl_txn_commit_block(struct ovsdb_idl_txn *txn)
+{
+    enum ovsdb_idl_txn_status status;
+
+    fatal_signal_run();
+    while ((status = ovsdb_idl_txn_commit(txn)) == TXN_INCOMPLETE) {
+        ovsdb_idl_run(txn->idl);
+        ovsdb_idl_wait(txn->idl);
+        ovsdb_idl_txn_wait(txn);
+        poll_block();
+    }
+    return status;
+}
+
 int64_t
 ovsdb_idl_txn_get_increment_new_value(const struct ovsdb_idl_txn *txn)
 {
@@ -1161,6 +1308,52 @@ ovsdb_idl_txn_abort(struct ovsdb_idl_txn *txn)
     }
 }
 
+const char *
+ovsdb_idl_txn_get_error(const struct ovsdb_idl_txn *txn)
+{
+    if (txn->status != TXN_ERROR) {
+        return ovsdb_idl_txn_status_to_string(txn->status);
+    } else if (txn->error) {
+        return txn->error;
+    } else {
+        return "no error details available";
+    }
+}
+
+static void
+ovsdb_idl_txn_set_error_json(struct ovsdb_idl_txn *txn,
+                             const struct json *json)
+{
+    if (txn->error == NULL) {
+        txn->error = json_to_string(json, JSSF_SORT);
+    }
+}
+
+/* For transaction 'txn' that completed successfully, finds and returns the
+ * permanent UUID that the database assigned to a newly inserted row, given the
+ * 'uuid' that ovsdb_idl_txn_insert() assigned locally to that row.
+ *
+ * Returns NULL if 'uuid' is not a UUID assigned by ovsdb_idl_txn_insert() or
+ * if it was assigned by that function and then deleted by
+ * ovsdb_idl_txn_delete() within the same transaction.  (Rows that are inserted
+ * and then deleted within a single transaction are never sent to the database
+ * server, so it never assigns them a permanent UUID.) */
+const struct uuid *
+ovsdb_idl_txn_get_insert_uuid(const struct ovsdb_idl_txn *txn,
+                              const struct uuid *uuid)
+{
+    const struct ovsdb_idl_txn_insert *insert;
+
+    assert(txn->status == TXN_SUCCESS || txn->status == TXN_UNCHANGED);
+    HMAP_FOR_EACH_IN_BUCKET (insert, struct ovsdb_idl_txn_insert, hmap_node,
+                             uuid_hash(uuid), &txn->inserted_rows) {
+        if (uuid_equals(uuid, &insert->dummy)) {
+            return &insert->real;
+        }
+    }
+    return NULL;
+}
+
 static void
 ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
                        enum ovsdb_idl_txn_status status)
@@ -1170,14 +1363,34 @@ ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
 }
 
 void
-ovsdb_idl_txn_write(struct ovsdb_idl_row *row,
+ovsdb_idl_txn_read(const struct ovsdb_idl_row *row,
+                   const struct ovsdb_idl_column *column,
+                   struct ovsdb_datum *datum)
+{
+    const struct ovsdb_idl_table_class *class = row->table->class;
+    size_t column_idx = column - class->columns;
+
+    assert(row->new != NULL);
+    if (row->written && bitmap_is_set(row->written, column_idx)) {
+        ovsdb_datum_clone(datum, &row->new[column_idx], &column->type);
+    } else if (row->old) {
+        ovsdb_datum_clone(datum, &row->old[column_idx], &column->type);
+    } else {
+        ovsdb_datum_init_default(datum, &column->type);
+    }
+}
+
+void
+ovsdb_idl_txn_write(const struct ovsdb_idl_row *row_,
                     const struct ovsdb_idl_column *column,
                     struct ovsdb_datum *datum)
 {
+    struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
     const struct ovsdb_idl_table_class *class = row->table->class;
     size_t column_idx = column - class->columns;
 
-    assert(row->new);
+    assert(row->new != NULL);
+    assert(column_idx < class->n_columns);
     if (hmap_node_is_null(&row->txn_node)) {
         hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
                     uuid_hash(&row->uuid));
@@ -1194,6 +1407,8 @@ ovsdb_idl_txn_write(struct ovsdb_idl_row *row,
         bitmap_set1(row->written, column_idx);
     }
     row->new[column_idx] = *datum;
+    (column->unparse)(row);
+    (column->parse)(row, &row->new[column_idx]);
 }
 
 void
@@ -1204,7 +1419,7 @@ ovsdb_idl_txn_verify(const struct ovsdb_idl_row *row_,
     const struct ovsdb_idl_table_class *class = row->table->class;
     size_t column_idx = column - class->columns;
 
-    assert(row->new);
+    assert(row->new != NULL);
     if (!row->old
         || (row->written && bitmap_is_set(row->written, column_idx))) {
         return;
@@ -1221,12 +1436,16 @@ ovsdb_idl_txn_verify(const struct ovsdb_idl_row *row_,
 }
 
 void
-ovsdb_idl_txn_delete(struct ovsdb_idl_row *row)
+ovsdb_idl_txn_delete(const struct ovsdb_idl_row *row_)
 {
-    assert(row->new);
+    struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
+
+    assert(row->new != NULL);
     if (!row->old) {
+        ovsdb_idl_row_unparse(row);
         ovsdb_idl_row_clear_new(row);
         assert(!row->prereqs);
+        hmap_remove(&row->table->rows, &row->hmap_node);
         hmap_remove(&row->table->idl->txn->txn_rows, &row->txn_node);
         free(row);
         return;
@@ -1239,7 +1458,7 @@ ovsdb_idl_txn_delete(struct ovsdb_idl_row *row)
     row->new = NULL;
 }
 
-struct ovsdb_idl_row *
+const struct ovsdb_idl_row *
 ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
                      const struct ovsdb_idl_table_class *class)
 {
@@ -1248,6 +1467,7 @@ ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
     row->table = ovsdb_idl_table_from_class(txn->idl, class);
     row->new = xmalloc(class->n_columns * sizeof *row->new);
     row->written = bitmap_allocate(class->n_columns);
+    hmap_insert(&row->table->rows, &row->hmap_node, uuid_hash(&row->uuid));
     hmap_insert(&txn->txn_rows, &row->txn_node, uuid_hash(&row->uuid));
     return row;
 }
@@ -1307,7 +1527,7 @@ ovsdb_idl_txn_process_inc_reply(struct ovsdb_idl_txn *txn,
         return false;
     }
 
-    /* We know that this is a JSON objects because the loop in
+    /* We know that this is a JSON object because the loop in
      * ovsdb_idl_txn_process_reply() checked. */
     mutate = json_object(results->elems[txn->inc_index]);
     count = shash_find_data(mutate, "count");
@@ -1345,6 +1565,44 @@ ovsdb_idl_txn_process_inc_reply(struct ovsdb_idl_txn *txn,
     return true;
 }
 
+static bool
+ovsdb_idl_txn_process_insert_reply(struct ovsdb_idl_txn_insert *insert,
+                                   const struct json_array *results)
+{
+    static const struct ovsdb_base_type uuid_type = OVSDB_BASE_UUID_INIT;
+    struct ovsdb_error *error;
+    struct json *json_uuid;
+    union ovsdb_atom uuid;
+    struct shash *reply;
+
+    if (insert->op_index >= results->n) {
+        VLOG_WARN_RL(&syntax_rl, "reply does not contain enough operations "
+                     "for insert (has %u, needs %u)",
+                     results->n, insert->op_index);
+        return false;
+    }
+
+    /* We know that this is a JSON object because the loop in
+     * ovsdb_idl_txn_process_reply() checked. */
+    reply = json_object(results->elems[insert->op_index]);
+    json_uuid = shash_find_data(reply, "uuid");
+    if (!check_json_type(json_uuid, JSON_ARRAY, "\"insert\" reply \"uuid\"")) {
+        return false;
+    }
+
+    error = ovsdb_atom_from_json(&uuid, &uuid_type, json_uuid, NULL);
+    if (error) {
+        char *s = ovsdb_error_to_string(error);
+        VLOG_WARN_RL(&syntax_rl, "\"insert\" reply \"uuid\" is not a JSON "
+                     "UUID: %s", s);
+        free(s);
+        return false;
+    }
+
+    insert->real = uuid.uuid;
+
+    return true;
+}
 
 static bool
 ovsdb_idl_txn_process_reply(struct ovsdb_idl *idl,
@@ -1364,47 +1622,58 @@ ovsdb_idl_txn_process_reply(struct ovsdb_idl *idl,
         VLOG_WARN_RL(&syntax_rl, "reply to \"transact\" is not JSON array");
         status = TXN_ERROR;
     } else {
+        struct json_array *ops = &msg->result->u.array;
         int hard_errors = 0;
         int soft_errors = 0;
         size_t i;
 
-        for (i = 0; i < msg->result->u.array.n; i++) {
-            struct json *json = msg->result->u.array.elems[i];
+        for (i = 0; i < ops->n; i++) {
+            struct json *op = ops->elems[i];
 
-            if (json->type == JSON_NULL) {
+            if (op->type == JSON_NULL) {
                 /* This isn't an error in itself but indicates that some prior
                  * operation failed, so make sure that we know about it. */
                 soft_errors++;
-            } else if (json->type == JSON_OBJECT) {
+            } else if (op->type == JSON_OBJECT) {
                 struct json *error;
 
-                error = shash_find_data(json_object(json), "error");
+                error = shash_find_data(json_object(op), "error");
                 if (error) {
                     if (error->type == JSON_STRING) {
                         if (!strcmp(error->u.string, "timed out")) {
                             soft_errors++;
                         } else if (strcmp(error->u.string, "aborted")) {
                             hard_errors++;
+                            ovsdb_idl_txn_set_error_json(txn, op);
                         }
                     } else {
                         hard_errors++;
+                        ovsdb_idl_txn_set_error_json(txn, op);
                         VLOG_WARN_RL(&syntax_rl,
                                      "\"error\" in reply is not JSON string");
                     }
                 }
             } else {
                 hard_errors++;
+                ovsdb_idl_txn_set_error_json(txn, op);
                 VLOG_WARN_RL(&syntax_rl,
                              "operation reply is not JSON null or object");
             }
         }
 
-        if (txn->inc_table
-            && !soft_errors
-            && !hard_errors
-            && !ovsdb_idl_txn_process_inc_reply(txn,
-                                                json_array(msg->result))) {
-            hard_errors++;
+        if (!soft_errors && !hard_errors) {
+            struct ovsdb_idl_txn_insert *insert;
+
+            if (txn->inc_table && !ovsdb_idl_txn_process_inc_reply(txn, ops)) {
+                hard_errors++;
+            }
+
+            HMAP_FOR_EACH (insert, struct ovsdb_idl_txn_insert, hmap_node,
+                           &txn->inserted_rows) {
+                if (!ovsdb_idl_txn_process_insert_reply(insert, ops)) {
+                    hard_errors++;
+                }
+            }
         }
 
         status = (hard_errors ? TXN_ERROR
@@ -1423,3 +1692,10 @@ ovsdb_idl_txn_get(const struct ovsdb_idl_row *row)
     assert(txn != NULL);
     return txn;
 }
+
+struct ovsdb_idl *
+ovsdb_idl_txn_get_idl (struct ovsdb_idl_txn *txn)
+{
+    return txn->idl;
+}
+