bridge: Tolerate missing Port and Interface records for local port.
[openvswitch] / lib / ovsdb-idl.c
1 /* Copyright (c) 2009, 2010, 2011 Nicira Networks.
2  *
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at:
6  *
7  *     http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15
16 #include <config.h>
17
18 #include "ovsdb-idl.h"
19
20 #include <assert.h>
21 #include <errno.h>
22 #include <inttypes.h>
23 #include <limits.h>
24 #include <stdlib.h>
25
26 #include "bitmap.h"
27 #include "dynamic-string.h"
28 #include "fatal-signal.h"
29 #include "json.h"
30 #include "jsonrpc.h"
31 #include "ovsdb-data.h"
32 #include "ovsdb-error.h"
33 #include "ovsdb-idl-provider.h"
34 #include "poll-loop.h"
35 #include "shash.h"
36 #include "util.h"
37 #include "vlog.h"
38
39 VLOG_DEFINE_THIS_MODULE(ovsdb_idl);
40
41 /* An arc from one idl_row to another.  When row A contains a UUID that
42  * references row B, this is represented by an arc from A (the source) to B
43  * (the destination).
44  *
45  * Arcs from a row to itself are omitted, that is, src and dst are always
46  * different.
47  *
48  * Arcs are never duplicated, that is, even if there are multiple references
49  * from A to B, there is only a single arc from A to B.
50  *
51  * Arcs are directed: an arc from A to B is the converse of an an arc from B to
52  * A.  Both an arc and its converse may both be present, if each row refers
53  * to the other circularly.
54  *
55  * The source and destination row may be in the same table or in different
56  * tables.
57  */
58 struct ovsdb_idl_arc {
59     struct list src_node;       /* In src->src_arcs list. */
60     struct list dst_node;       /* In dst->dst_arcs list. */
61     struct ovsdb_idl_row *src;  /* Source row. */
62     struct ovsdb_idl_row *dst;  /* Destination row. */
63 };
64
65 struct ovsdb_idl {
66     const struct ovsdb_idl_class *class;
67     struct jsonrpc_session *session;
68     struct shash table_by_name;
69     struct ovsdb_idl_table *tables; /* Contains "struct ovsdb_idl_table *"s.*/
70     struct json *monitor_request_id;
71     unsigned int last_monitor_request_seqno;
72     unsigned int change_seqno;
73
74     /* Transaction support. */
75     struct ovsdb_idl_txn *txn;
76     struct hmap outstanding_txns;
77 };
78
79 struct ovsdb_idl_txn {
80     struct hmap_node hmap_node;
81     struct json *request_id;
82     struct ovsdb_idl *idl;
83     struct hmap txn_rows;
84     enum ovsdb_idl_txn_status status;
85     char *error;
86     bool dry_run;
87     struct ds comment;
88
89     /* Increments. */
90     char *inc_table;
91     char *inc_column;
92     struct json *inc_where;
93     unsigned int inc_index;
94     int64_t inc_new_value;
95
96     /* Inserted rows. */
97     struct hmap inserted_rows;
98 };
99
100 struct ovsdb_idl_txn_insert {
101     struct hmap_node hmap_node; /* In struct ovsdb_idl_txn's inserted_rows. */
102     struct uuid dummy;          /* Dummy UUID used locally. */
103     int op_index;               /* Index into transaction's operation array. */
104     struct uuid real;           /* Real UUID used by database server. */
105 };
106
107 static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);
108 static struct vlog_rate_limit semantic_rl = VLOG_RATE_LIMIT_INIT(1, 5);
109
110 static void ovsdb_idl_clear(struct ovsdb_idl *);
111 static void ovsdb_idl_send_monitor_request(struct ovsdb_idl *);
112 static void ovsdb_idl_parse_update(struct ovsdb_idl *, const struct json *);
113 static struct ovsdb_error *ovsdb_idl_parse_update__(struct ovsdb_idl *,
114                                                     const struct json *);
115 static bool ovsdb_idl_process_update(struct ovsdb_idl_table *,
116                                      const struct uuid *,
117                                      const struct json *old,
118                                      const struct json *new);
119 static void ovsdb_idl_insert_row(struct ovsdb_idl_row *, const struct json *);
120 static void ovsdb_idl_delete_row(struct ovsdb_idl_row *);
121 static bool ovsdb_idl_modify_row(struct ovsdb_idl_row *, const struct json *);
122
123 static bool ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *);
124 static struct ovsdb_idl_row *ovsdb_idl_row_create__(
125     const struct ovsdb_idl_table_class *);
126 static struct ovsdb_idl_row *ovsdb_idl_row_create(struct ovsdb_idl_table *,
127                                                   const struct uuid *);
128 static void ovsdb_idl_row_destroy(struct ovsdb_idl_row *);
129
130 static void ovsdb_idl_row_parse(struct ovsdb_idl_row *);
131 static void ovsdb_idl_row_unparse(struct ovsdb_idl_row *);
132 static void ovsdb_idl_row_clear_old(struct ovsdb_idl_row *);
133 static void ovsdb_idl_row_clear_new(struct ovsdb_idl_row *);
134
135 static void ovsdb_idl_txn_abort_all(struct ovsdb_idl *);
136 static bool ovsdb_idl_txn_process_reply(struct ovsdb_idl *,
137                                         const struct jsonrpc_msg *msg);
138
139 /* Creates and returns a connection to database 'remote', which should be in a
140  * form acceptable to jsonrpc_session_open().  The connection will maintain an
141  * in-memory replica of the remote database whose schema is described by
142  * 'class'.  (Ordinarily 'class' is compiled from an OVSDB schema automatically
143  * by ovsdb-idlc.)
144  *
145  * If 'monitor_everything_by_default' is true, then everything in the remote
146  * database will be replicated by default.  ovsdb_idl_omit() and
147  * ovsdb_idl_omit_alert() may be used to selectively drop some columns from
148  * monitoring.
149  *
150  * If 'monitor_everything_by_default' is false, then no columns or tables will
151  * be replicated by default.  ovsdb_idl_add_column() and ovsdb_idl_add_table()
152  * must be used to choose some columns or tables to replicate.
153  */
154 struct ovsdb_idl *
155 ovsdb_idl_create(const char *remote, const struct ovsdb_idl_class *class,
156                  bool monitor_everything_by_default)
157 {
158     struct ovsdb_idl *idl;
159     uint8_t default_mode;
160     size_t i;
161
162     default_mode = (monitor_everything_by_default
163                     ? OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT
164                     : 0);
165
166     idl = xzalloc(sizeof *idl);
167     idl->class = class;
168     idl->session = jsonrpc_session_open(remote);
169     shash_init(&idl->table_by_name);
170     idl->tables = xmalloc(class->n_tables * sizeof *idl->tables);
171     for (i = 0; i < class->n_tables; i++) {
172         const struct ovsdb_idl_table_class *tc = &class->tables[i];
173         struct ovsdb_idl_table *table = &idl->tables[i];
174         size_t j;
175
176         shash_add_assert(&idl->table_by_name, tc->name, table);
177         table->class = tc;
178         table->modes = xmalloc(tc->n_columns);
179         memset(table->modes, default_mode, tc->n_columns);
180         table->need_table = false;
181         shash_init(&table->columns);
182         for (j = 0; j < tc->n_columns; j++) {
183             const struct ovsdb_idl_column *column = &tc->columns[j];
184
185             shash_add_assert(&table->columns, column->name, column);
186         }
187         hmap_init(&table->rows);
188         table->idl = idl;
189     }
190     idl->last_monitor_request_seqno = UINT_MAX;
191     hmap_init(&idl->outstanding_txns);
192
193     return idl;
194 }
195
196 /* Destroys 'idl' and all of the data structures that it manages. */
197 void
198 ovsdb_idl_destroy(struct ovsdb_idl *idl)
199 {
200     if (idl) {
201         size_t i;
202
203         assert(!idl->txn);
204         ovsdb_idl_clear(idl);
205         jsonrpc_session_close(idl->session);
206
207         for (i = 0; i < idl->class->n_tables; i++) {
208             struct ovsdb_idl_table *table = &idl->tables[i];
209             shash_destroy(&table->columns);
210             hmap_destroy(&table->rows);
211             free(table->modes);
212         }
213         shash_destroy(&idl->table_by_name);
214         free(idl->tables);
215         json_destroy(idl->monitor_request_id);
216         free(idl);
217     }
218 }
219
220 static void
221 ovsdb_idl_clear(struct ovsdb_idl *idl)
222 {
223     bool changed = false;
224     size_t i;
225
226     for (i = 0; i < idl->class->n_tables; i++) {
227         struct ovsdb_idl_table *table = &idl->tables[i];
228         struct ovsdb_idl_row *row, *next_row;
229
230         if (hmap_is_empty(&table->rows)) {
231             continue;
232         }
233
234         changed = true;
235         HMAP_FOR_EACH_SAFE (row, next_row, hmap_node, &table->rows) {
236             struct ovsdb_idl_arc *arc, *next_arc;
237
238             if (!ovsdb_idl_row_is_orphan(row)) {
239                 ovsdb_idl_row_unparse(row);
240             }
241             LIST_FOR_EACH_SAFE (arc, next_arc, src_node, &row->src_arcs) {
242                 free(arc);
243             }
244             /* No need to do anything with dst_arcs: some node has those arcs
245              * as forward arcs and will destroy them itself. */
246
247             ovsdb_idl_row_destroy(row);
248         }
249     }
250
251     if (changed) {
252         idl->change_seqno++;
253     }
254 }
255
256 /* Processes a batch of messages from the database server on 'idl'.  Returns
257  * true if the database as seen through 'idl' changed, false if it did not
258  * change.  The initial fetch of the entire contents of the remote database is
259  * considered to be one kind of change.
260  *
261  * When this function returns false, the client may continue to use any data
262  * structures it obtained from 'idl' in the past.  But when it returns true,
263  * the client must not access any of these data structures again, because they
264  * could have freed or reused for other purposes.
265  *
266  * This function can return occasional false positives, that is, report that
267  * the database changed even though it didn't.  This happens if the connection
268  * to the database drops and reconnects, which causes the database contents to
269  * be reloaded even if they didn't change.  (It could also happen if the
270  * database server sends out a "change" that reflects what we already thought
271  * was in the database, but the database server is not supposed to do that.)
272  *
273  * As an alternative to checking the return value, the client may check for
274  * changes in the value returned by ovsdb_idl_get_seqno().
275  */
276 bool
277 ovsdb_idl_run(struct ovsdb_idl *idl)
278 {
279     unsigned int initial_change_seqno = idl->change_seqno;
280     int i;
281
282     assert(!idl->txn);
283     jsonrpc_session_run(idl->session);
284     for (i = 0; jsonrpc_session_is_connected(idl->session) && i < 50; i++) {
285         struct jsonrpc_msg *msg;
286         unsigned int seqno;
287
288         seqno = jsonrpc_session_get_seqno(idl->session);
289         if (idl->last_monitor_request_seqno != seqno) {
290             idl->last_monitor_request_seqno = seqno;
291             ovsdb_idl_txn_abort_all(idl);
292             ovsdb_idl_send_monitor_request(idl);
293             break;
294         }
295
296         msg = jsonrpc_session_recv(idl->session);
297         if (!msg) {
298             break;
299         }
300
301         if (msg->type == JSONRPC_NOTIFY
302                    && !strcmp(msg->method, "update")
303                    && msg->params->type == JSON_ARRAY
304                    && msg->params->u.array.n == 2
305                    && msg->params->u.array.elems[0]->type == JSON_NULL) {
306             ovsdb_idl_parse_update(idl, msg->params->u.array.elems[1]);
307         } else if (msg->type == JSONRPC_REPLY
308                    && idl->monitor_request_id
309                    && json_equal(idl->monitor_request_id, msg->id)) {
310             idl->change_seqno++;
311             json_destroy(idl->monitor_request_id);
312             idl->monitor_request_id = NULL;
313             ovsdb_idl_clear(idl);
314             ovsdb_idl_parse_update(idl, msg->result);
315         } else if (msg->type == JSONRPC_REPLY && msg->id->type == JSON_STRING
316                    && !strcmp(msg->id->u.string, "echo")) {
317             /* It's a reply to our echo request.  Ignore it. */
318         } else if ((msg->type == JSONRPC_ERROR
319                     || msg->type == JSONRPC_REPLY)
320                    && ovsdb_idl_txn_process_reply(idl, msg)) {
321             /* ovsdb_idl_txn_process_reply() did everything needful. */
322         } else {
323             /* This can happen if ovsdb_idl_txn_destroy() is called to destroy
324              * a transaction before we receive the reply, so keep the log level
325              * low. */
326             VLOG_DBG("%s: received unexpected %s message",
327                      jsonrpc_session_get_name(idl->session),
328                      jsonrpc_msg_type_to_string(msg->type));
329         }
330         jsonrpc_msg_destroy(msg);
331     }
332
333     return initial_change_seqno != idl->change_seqno;
334 }
335
336 /* Arranges for poll_block() to wake up when ovsdb_idl_run() has something to
337  * do or when activity occurs on a transaction on 'idl'. */
338 void
339 ovsdb_idl_wait(struct ovsdb_idl *idl)
340 {
341     jsonrpc_session_wait(idl->session);
342     jsonrpc_session_recv_wait(idl->session);
343 }
344
345 /* Returns a number that represents the state of 'idl'.  When 'idl' is updated
346  * (by ovsdb_idl_run()), the return value changes. */
347 unsigned int
348 ovsdb_idl_get_seqno(const struct ovsdb_idl *idl)
349 {
350     return idl->change_seqno;
351 }
352
353 /* Returns true if 'idl' successfully connected to the remote database and
354  * retrieved its contents (even if the connection subsequently dropped and is
355  * in the process of reconnecting).  If so, then 'idl' contains an atomic
356  * snapshot of the database's contents (but it might be arbitrarily old if the
357  * connection dropped).
358  *
359  * Returns false if 'idl' has never connected or retrieved the database's
360  * contents.  If so, 'idl' is empty. */
361 bool
362 ovsdb_idl_has_ever_connected(const struct ovsdb_idl *idl)
363 {
364     return ovsdb_idl_get_seqno(idl) != 0;
365 }
366
367 /* Forces 'idl' to drop its connection to the database and reconnect.  In the
368  * meantime, the contents of 'idl' will not change. */
369 void
370 ovsdb_idl_force_reconnect(struct ovsdb_idl *idl)
371 {
372     jsonrpc_session_force_reconnect(idl->session);
373 }
374 \f
375 static unsigned char *
376 ovsdb_idl_get_mode(struct ovsdb_idl *idl,
377                    const struct ovsdb_idl_column *column)
378 {
379     size_t i;
380
381     assert(!idl->change_seqno);
382
383     for (i = 0; i < idl->class->n_tables; i++) {
384         const struct ovsdb_idl_table *table = &idl->tables[i];
385         const struct ovsdb_idl_table_class *tc = table->class;
386
387         if (column >= tc->columns && column < &tc->columns[tc->n_columns]) {
388             return &table->modes[column - tc->columns];
389         }
390     }
391
392     NOT_REACHED();
393 }
394
395 static void
396 add_ref_table(struct ovsdb_idl *idl, const struct ovsdb_base_type *base)
397 {
398     if (base->type == OVSDB_TYPE_UUID && base->u.uuid.refTableName) {
399         struct ovsdb_idl_table *table;
400
401         table = shash_find_data(&idl->table_by_name,
402                                 base->u.uuid.refTableName);
403         if (table) {
404             table->need_table = true;
405         } else {
406             VLOG_WARN("%s IDL class missing referenced table %s",
407                       idl->class->database, base->u.uuid.refTableName);
408         }
409     }
410 }
411
412 /* Turns on OVSDB_IDL_MONITOR and OVSDB_IDL_ALERT for 'column' in 'idl'.  Also
413  * ensures that any tables referenced by 'column' will be replicated, even if
414  * no columns in that table are selected for replication (see
415  * ovsdb_idl_add_table() for more information).
416  *
417  * This function is only useful if 'monitor_everything_by_default' was false in
418  * the call to ovsdb_idl_create().  This function should be called between
419  * ovsdb_idl_create() and the first call to ovsdb_idl_run().
420  */
421 void
422 ovsdb_idl_add_column(struct ovsdb_idl *idl,
423                      const struct ovsdb_idl_column *column)
424 {
425     *ovsdb_idl_get_mode(idl, column) = OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT;
426     add_ref_table(idl, &column->type.key);
427     add_ref_table(idl, &column->type.value);
428 }
429
430 /* Ensures that the table with class 'tc' will be replicated on 'idl' even if
431  * no columns are selected for replication.  This can be useful because it
432  * allows 'idl' to keep track of what rows in the table actually exist, which
433  * in turn allows columns that reference the table to have accurate contents.
434  * (The IDL presents the database with references to rows that do not exist
435  * removed.)
436  *
437  * This function is only useful if 'monitor_everything_by_default' was false in
438  * the call to ovsdb_idl_create().  This function should be called between
439  * ovsdb_idl_create() and the first call to ovsdb_idl_run().
440  */
441 void
442 ovsdb_idl_add_table(struct ovsdb_idl *idl,
443                     const struct ovsdb_idl_table_class *tc)
444 {
445     size_t i;
446
447     for (i = 0; i < idl->class->n_tables; i++) {
448         struct ovsdb_idl_table *table = &idl->tables[i];
449
450         if (table->class == tc) {
451             table->need_table = true;
452             return;
453         }
454     }
455
456     NOT_REACHED();
457 }
458
459 /* Turns off OVSDB_IDL_ALERT for 'column' in 'idl'.
460  *
461  * This function should be called between ovsdb_idl_create() and the first call
462  * to ovsdb_idl_run().
463  */
464 void
465 ovsdb_idl_omit_alert(struct ovsdb_idl *idl,
466                      const struct ovsdb_idl_column *column)
467 {
468     *ovsdb_idl_get_mode(idl, column) &= ~OVSDB_IDL_ALERT;
469 }
470
471 /* Sets the mode for 'column' in 'idl' to 0.  See the big comment above
472  * OVSDB_IDL_MONITOR for details.
473  *
474  * This function should be called between ovsdb_idl_create() and the first call
475  * to ovsdb_idl_run().
476  */
477 void
478 ovsdb_idl_omit(struct ovsdb_idl *idl, const struct ovsdb_idl_column *column)
479 {
480     *ovsdb_idl_get_mode(idl, column) = 0;
481 }
482 \f
483 static void
484 ovsdb_idl_send_monitor_request(struct ovsdb_idl *idl)
485 {
486     struct json *monitor_requests;
487     struct jsonrpc_msg *msg;
488     size_t i;
489
490     monitor_requests = json_object_create();
491     for (i = 0; i < idl->class->n_tables; i++) {
492         const struct ovsdb_idl_table *table = &idl->tables[i];
493         const struct ovsdb_idl_table_class *tc = table->class;
494         struct json *monitor_request, *columns;
495         size_t j;
496
497         columns = table->need_table ? json_array_create_empty() : NULL;
498         for (j = 0; j < tc->n_columns; j++) {
499             const struct ovsdb_idl_column *column = &tc->columns[j];
500             if (table->modes[j] & OVSDB_IDL_MONITOR) {
501                 if (!columns) {
502                     columns = json_array_create_empty();
503                 }
504                 json_array_add(columns, json_string_create(column->name));
505             }
506         }
507
508         if (columns) {
509             monitor_request = json_object_create();
510             json_object_put(monitor_request, "columns", columns);
511             json_object_put(monitor_requests, tc->name, monitor_request);
512         }
513     }
514
515     json_destroy(idl->monitor_request_id);
516     msg = jsonrpc_create_request(
517         "monitor",
518         json_array_create_3(json_string_create(idl->class->database),
519                             json_null_create(), monitor_requests),
520         &idl->monitor_request_id);
521     jsonrpc_session_send(idl->session, msg);
522 }
523
524 static void
525 ovsdb_idl_parse_update(struct ovsdb_idl *idl, const struct json *table_updates)
526 {
527     struct ovsdb_error *error = ovsdb_idl_parse_update__(idl, table_updates);
528     if (error) {
529         if (!VLOG_DROP_WARN(&syntax_rl)) {
530             char *s = ovsdb_error_to_string(error);
531             VLOG_WARN_RL(&syntax_rl, "%s", s);
532             free(s);
533         }
534         ovsdb_error_destroy(error);
535     }
536 }
537
538 static struct ovsdb_error *
539 ovsdb_idl_parse_update__(struct ovsdb_idl *idl,
540                          const struct json *table_updates)
541 {
542     const struct shash_node *tables_node;
543
544     if (table_updates->type != JSON_OBJECT) {
545         return ovsdb_syntax_error(table_updates, NULL,
546                                   "<table-updates> is not an object");
547     }
548     SHASH_FOR_EACH (tables_node, json_object(table_updates)) {
549         const struct json *table_update = tables_node->data;
550         const struct shash_node *table_node;
551         struct ovsdb_idl_table *table;
552
553         table = shash_find_data(&idl->table_by_name, tables_node->name);
554         if (!table) {
555             return ovsdb_syntax_error(
556                 table_updates, NULL,
557                 "<table-updates> includes unknown table \"%s\"",
558                 tables_node->name);
559         }
560
561         if (table_update->type != JSON_OBJECT) {
562             return ovsdb_syntax_error(table_update, NULL,
563                                       "<table-update> for table \"%s\" is "
564                                       "not an object", table->class->name);
565         }
566         SHASH_FOR_EACH (table_node, json_object(table_update)) {
567             const struct json *row_update = table_node->data;
568             const struct json *old_json, *new_json;
569             struct uuid uuid;
570
571             if (!uuid_from_string(&uuid, table_node->name)) {
572                 return ovsdb_syntax_error(table_update, NULL,
573                                           "<table-update> for table \"%s\" "
574                                           "contains bad UUID "
575                                           "\"%s\" as member name",
576                                           table->class->name,
577                                           table_node->name);
578             }
579             if (row_update->type != JSON_OBJECT) {
580                 return ovsdb_syntax_error(row_update, NULL,
581                                           "<table-update> for table \"%s\" "
582                                           "contains <row-update> for %s that "
583                                           "is not an object",
584                                           table->class->name,
585                                           table_node->name);
586             }
587
588             old_json = shash_find_data(json_object(row_update), "old");
589             new_json = shash_find_data(json_object(row_update), "new");
590             if (old_json && old_json->type != JSON_OBJECT) {
591                 return ovsdb_syntax_error(old_json, NULL,
592                                           "\"old\" <row> is not object");
593             } else if (new_json && new_json->type != JSON_OBJECT) {
594                 return ovsdb_syntax_error(new_json, NULL,
595                                           "\"new\" <row> is not object");
596             } else if ((old_json != NULL) + (new_json != NULL)
597                        != shash_count(json_object(row_update))) {
598                 return ovsdb_syntax_error(row_update, NULL,
599                                           "<row-update> contains unexpected "
600                                           "member");
601             } else if (!old_json && !new_json) {
602                 return ovsdb_syntax_error(row_update, NULL,
603                                           "<row-update> missing \"old\" "
604                                           "and \"new\" members");
605             }
606
607             if (ovsdb_idl_process_update(table, &uuid, old_json, new_json)) {
608                 idl->change_seqno++;
609             }
610         }
611     }
612
613     return NULL;
614 }
615
616 static struct ovsdb_idl_row *
617 ovsdb_idl_get_row(struct ovsdb_idl_table *table, const struct uuid *uuid)
618 {
619     struct ovsdb_idl_row *row;
620
621     HMAP_FOR_EACH_WITH_HASH (row, hmap_node, uuid_hash(uuid), &table->rows) {
622         if (uuid_equals(&row->uuid, uuid)) {
623             return row;
624         }
625     }
626     return NULL;
627 }
628
629 /* Returns true if a column with mode OVSDB_IDL_MODE_RW changed, false
630  * otherwise. */
631 static bool
632 ovsdb_idl_process_update(struct ovsdb_idl_table *table,
633                          const struct uuid *uuid, const struct json *old,
634                          const struct json *new)
635 {
636     struct ovsdb_idl_row *row;
637
638     row = ovsdb_idl_get_row(table, uuid);
639     if (!new) {
640         /* Delete row. */
641         if (row && !ovsdb_idl_row_is_orphan(row)) {
642             /* XXX perhaps we should check the 'old' values? */
643             ovsdb_idl_delete_row(row);
644         } else {
645             VLOG_WARN_RL(&semantic_rl, "cannot delete missing row "UUID_FMT" "
646                          "from table %s",
647                          UUID_ARGS(uuid), table->class->name);
648             return false;
649         }
650     } else if (!old) {
651         /* Insert row. */
652         if (!row) {
653             ovsdb_idl_insert_row(ovsdb_idl_row_create(table, uuid), new);
654         } else if (ovsdb_idl_row_is_orphan(row)) {
655             ovsdb_idl_insert_row(row, new);
656         } else {
657             VLOG_WARN_RL(&semantic_rl, "cannot add existing row "UUID_FMT" to "
658                          "table %s", UUID_ARGS(uuid), table->class->name);
659             return ovsdb_idl_modify_row(row, new);
660         }
661     } else {
662         /* Modify row. */
663         if (row) {
664             /* XXX perhaps we should check the 'old' values? */
665             if (!ovsdb_idl_row_is_orphan(row)) {
666                 return ovsdb_idl_modify_row(row, new);
667             } else {
668                 VLOG_WARN_RL(&semantic_rl, "cannot modify missing but "
669                              "referenced row "UUID_FMT" in table %s",
670                              UUID_ARGS(uuid), table->class->name);
671                 ovsdb_idl_insert_row(row, new);
672             }
673         } else {
674             VLOG_WARN_RL(&semantic_rl, "cannot modify missing row "UUID_FMT" "
675                          "in table %s", UUID_ARGS(uuid), table->class->name);
676             ovsdb_idl_insert_row(ovsdb_idl_row_create(table, uuid), new);
677         }
678     }
679
680     return true;
681 }
682
683 /* Returns true if a column with mode OVSDB_IDL_MODE_RW changed, false
684  * otherwise. */
685 static bool
686 ovsdb_idl_row_update(struct ovsdb_idl_row *row, const struct json *row_json)
687 {
688     struct ovsdb_idl_table *table = row->table;
689     struct shash_node *node;
690     bool changed = false;
691
692     SHASH_FOR_EACH (node, json_object(row_json)) {
693         const char *column_name = node->name;
694         const struct ovsdb_idl_column *column;
695         struct ovsdb_datum datum;
696         struct ovsdb_error *error;
697
698         column = shash_find_data(&table->columns, column_name);
699         if (!column) {
700             VLOG_WARN_RL(&syntax_rl, "unknown column %s updating row "UUID_FMT,
701                          column_name, UUID_ARGS(&row->uuid));
702             continue;
703         }
704
705         error = ovsdb_datum_from_json(&datum, &column->type, node->data, NULL);
706         if (!error) {
707             unsigned int column_idx = column - table->class->columns;
708             struct ovsdb_datum *old = &row->old[column_idx];
709
710             if (!ovsdb_datum_equals(old, &datum, &column->type)) {
711                 ovsdb_datum_swap(old, &datum);
712                 if (table->modes[column_idx] & OVSDB_IDL_ALERT) {
713                     changed = true;
714                 }
715             } else {
716                 /* Didn't really change but the OVSDB monitor protocol always
717                  * includes every value in a row. */
718             }
719
720             ovsdb_datum_destroy(&datum, &column->type);
721         } else {
722             char *s = ovsdb_error_to_string(error);
723             VLOG_WARN_RL(&syntax_rl, "error parsing column %s in row "UUID_FMT
724                          " in table %s: %s", column_name,
725                          UUID_ARGS(&row->uuid), table->class->name, s);
726             free(s);
727             ovsdb_error_destroy(error);
728         }
729     }
730     return changed;
731 }
732
733 /* When a row A refers to row B through a column with a "refTable" constraint,
734  * but row B does not exist, row B is called an "orphan row".  Orphan rows
735  * should not persist, because the database enforces referential integrity, but
736  * they can appear transiently as changes from the database are received (the
737  * database doesn't try to topologically sort them and circular references mean
738  * it isn't always possible anyhow).
739  *
740  * This function returns true if 'row' is an orphan row, otherwise false.
741  */
742 static bool
743 ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *row)
744 {
745     return !row->old && !row->new;
746 }
747
748 /* Returns true if 'row' is conceptually part of the database as modified by
749  * the current transaction (if any), false otherwise.
750  *
751  * This function will return true if 'row' is not an orphan (see the comment on
752  * ovsdb_idl_row_is_orphan()) and:
753  *
754  *   - 'row' exists in the database and has not been deleted within the
755  *     current transaction (if any).
756  *
757  *   - 'row' was inserted within the current transaction and has not been
758  *     deleted.  (In the latter case you should not have passed 'row' in at
759  *     all, because ovsdb_idl_txn_delete() freed it.)
760  *
761  * This function will return false if 'row' is an orphan or if 'row' was
762  * deleted within the current transaction.
763  */
764 static bool
765 ovsdb_idl_row_exists(const struct ovsdb_idl_row *row)
766 {
767     return row->new != NULL;
768 }
769
770 static void
771 ovsdb_idl_row_parse(struct ovsdb_idl_row *row)
772 {
773     const struct ovsdb_idl_table_class *class = row->table->class;
774     size_t i;
775
776     for (i = 0; i < class->n_columns; i++) {
777         const struct ovsdb_idl_column *c = &class->columns[i];
778         (c->parse)(row, &row->old[i]);
779     }
780 }
781
782 static void
783 ovsdb_idl_row_unparse(struct ovsdb_idl_row *row)
784 {
785     const struct ovsdb_idl_table_class *class = row->table->class;
786     size_t i;
787
788     for (i = 0; i < class->n_columns; i++) {
789         const struct ovsdb_idl_column *c = &class->columns[i];
790         (c->unparse)(row);
791     }
792 }
793
794 static void
795 ovsdb_idl_row_clear_old(struct ovsdb_idl_row *row)
796 {
797     assert(row->old == row->new);
798     if (!ovsdb_idl_row_is_orphan(row)) {
799         const struct ovsdb_idl_table_class *class = row->table->class;
800         size_t i;
801
802         for (i = 0; i < class->n_columns; i++) {
803             ovsdb_datum_destroy(&row->old[i], &class->columns[i].type);
804         }
805         free(row->old);
806         row->old = row->new = NULL;
807     }
808 }
809
810 static void
811 ovsdb_idl_row_clear_new(struct ovsdb_idl_row *row)
812 {
813     if (row->old != row->new) {
814         if (row->new) {
815             const struct ovsdb_idl_table_class *class = row->table->class;
816             size_t i;
817
818             if (row->written) {
819                 BITMAP_FOR_EACH_1 (i, class->n_columns, row->written) {
820                     ovsdb_datum_destroy(&row->new[i], &class->columns[i].type);
821                 }
822             }
823             free(row->new);
824             free(row->written);
825             row->written = NULL;
826         }
827         row->new = row->old;
828     }
829 }
830
831 static void
832 ovsdb_idl_row_clear_arcs(struct ovsdb_idl_row *row, bool destroy_dsts)
833 {
834     struct ovsdb_idl_arc *arc, *next;
835
836     /* Delete all forward arcs.  If 'destroy_dsts', destroy any orphaned rows
837      * that this causes to be unreferenced. */
838     LIST_FOR_EACH_SAFE (arc, next, src_node, &row->src_arcs) {
839         list_remove(&arc->dst_node);
840         if (destroy_dsts
841             && ovsdb_idl_row_is_orphan(arc->dst)
842             && list_is_empty(&arc->dst->dst_arcs)) {
843             ovsdb_idl_row_destroy(arc->dst);
844         }
845         free(arc);
846     }
847     list_init(&row->src_arcs);
848 }
849
850 /* Force nodes that reference 'row' to reparse. */
851 static void
852 ovsdb_idl_row_reparse_backrefs(struct ovsdb_idl_row *row)
853 {
854     struct ovsdb_idl_arc *arc, *next;
855
856     /* This is trickier than it looks.  ovsdb_idl_row_clear_arcs() will destroy
857      * 'arc', so we need to use the "safe" variant of list traversal.  However,
858      * calling an ovsdb_idl_column's 'parse' function will add an arc
859      * equivalent to 'arc' to row->arcs.  That could be a problem for
860      * traversal, but it adds it at the beginning of the list to prevent us
861      * from stumbling upon it again.
862      *
863      * (If duplicate arcs were possible then we would need to make sure that
864      * 'next' didn't also point into 'arc''s destination, but we forbid
865      * duplicate arcs.) */
866     LIST_FOR_EACH_SAFE (arc, next, dst_node, &row->dst_arcs) {
867         struct ovsdb_idl_row *ref = arc->src;
868
869         ovsdb_idl_row_unparse(ref);
870         ovsdb_idl_row_clear_arcs(ref, false);
871         ovsdb_idl_row_parse(ref);
872     }
873 }
874
875 static struct ovsdb_idl_row *
876 ovsdb_idl_row_create__(const struct ovsdb_idl_table_class *class)
877 {
878     struct ovsdb_idl_row *row = xzalloc(class->allocation_size);
879     list_init(&row->src_arcs);
880     list_init(&row->dst_arcs);
881     hmap_node_nullify(&row->txn_node);
882     return row;
883 }
884
885 static struct ovsdb_idl_row *
886 ovsdb_idl_row_create(struct ovsdb_idl_table *table, const struct uuid *uuid)
887 {
888     struct ovsdb_idl_row *row = ovsdb_idl_row_create__(table->class);
889     hmap_insert(&table->rows, &row->hmap_node, uuid_hash(uuid));
890     row->uuid = *uuid;
891     row->table = table;
892     return row;
893 }
894
895 static void
896 ovsdb_idl_row_destroy(struct ovsdb_idl_row *row)
897 {
898     if (row) {
899         ovsdb_idl_row_clear_old(row);
900         hmap_remove(&row->table->rows, &row->hmap_node);
901         free(row);
902     }
903 }
904
905 static void
906 ovsdb_idl_insert_row(struct ovsdb_idl_row *row, const struct json *row_json)
907 {
908     const struct ovsdb_idl_table_class *class = row->table->class;
909     size_t i;
910
911     assert(!row->old && !row->new);
912     row->old = row->new = xmalloc(class->n_columns * sizeof *row->old);
913     for (i = 0; i < class->n_columns; i++) {
914         ovsdb_datum_init_default(&row->old[i], &class->columns[i].type);
915     }
916     ovsdb_idl_row_update(row, row_json);
917     ovsdb_idl_row_parse(row);
918
919     ovsdb_idl_row_reparse_backrefs(row);
920 }
921
922 static void
923 ovsdb_idl_delete_row(struct ovsdb_idl_row *row)
924 {
925     ovsdb_idl_row_unparse(row);
926     ovsdb_idl_row_clear_arcs(row, true);
927     ovsdb_idl_row_clear_old(row);
928     if (list_is_empty(&row->dst_arcs)) {
929         ovsdb_idl_row_destroy(row);
930     } else {
931         ovsdb_idl_row_reparse_backrefs(row);
932     }
933 }
934
935 /* Returns true if a column with mode OVSDB_IDL_MODE_RW changed, false
936  * otherwise. */
937 static bool
938 ovsdb_idl_modify_row(struct ovsdb_idl_row *row, const struct json *row_json)
939 {
940     bool changed;
941
942     ovsdb_idl_row_unparse(row);
943     ovsdb_idl_row_clear_arcs(row, true);
944     changed = ovsdb_idl_row_update(row, row_json);
945     ovsdb_idl_row_parse(row);
946
947     return changed;
948 }
949
950 static bool
951 may_add_arc(const struct ovsdb_idl_row *src, const struct ovsdb_idl_row *dst)
952 {
953     const struct ovsdb_idl_arc *arc;
954
955     /* No self-arcs. */
956     if (src == dst) {
957         return false;
958     }
959
960     /* No duplicate arcs.
961      *
962      * We only need to test whether the first arc in dst->dst_arcs originates
963      * at 'src', since we add all of the arcs from a given source in a clump
964      * (in a single call to ovsdb_idl_row_parse()) and new arcs are always
965      * added at the front of the dst_arcs list. */
966     if (list_is_empty(&dst->dst_arcs)) {
967         return true;
968     }
969     arc = CONTAINER_OF(dst->dst_arcs.next, struct ovsdb_idl_arc, dst_node);
970     return arc->src != src;
971 }
972
973 static struct ovsdb_idl_table *
974 ovsdb_idl_table_from_class(const struct ovsdb_idl *idl,
975                            const struct ovsdb_idl_table_class *table_class)
976 {
977     return &idl->tables[table_class - idl->class->tables];
978 }
979
980 struct ovsdb_idl_row *
981 ovsdb_idl_get_row_arc(struct ovsdb_idl_row *src,
982                       struct ovsdb_idl_table_class *dst_table_class,
983                       const struct uuid *dst_uuid)
984 {
985     struct ovsdb_idl *idl = src->table->idl;
986     struct ovsdb_idl_table *dst_table;
987     struct ovsdb_idl_arc *arc;
988     struct ovsdb_idl_row *dst;
989
990     dst_table = ovsdb_idl_table_from_class(idl, dst_table_class);
991     dst = ovsdb_idl_get_row(dst_table, dst_uuid);
992     if (idl->txn) {
993         /* We're being called from ovsdb_idl_txn_write().  We must not update
994          * any arcs, because the transaction will be backed out at commit or
995          * abort time and we don't want our graph screwed up.
996          *
997          * Just return the destination row, if there is one and it has not been
998          * deleted. */
999         if (dst && (hmap_node_is_null(&dst->txn_node) || dst->new)) {
1000             return dst;
1001         }
1002         return NULL;
1003     } else {
1004         /* We're being called from some other context.  Update the graph. */
1005         if (!dst) {
1006             dst = ovsdb_idl_row_create(dst_table, dst_uuid);
1007         }
1008
1009         /* Add a new arc, if it wouldn't be a self-arc or a duplicate arc. */
1010         if (may_add_arc(src, dst)) {
1011             /* The arc *must* be added at the front of the dst_arcs list.  See
1012              * ovsdb_idl_row_reparse_backrefs() for details. */
1013             arc = xmalloc(sizeof *arc);
1014             list_push_front(&src->src_arcs, &arc->src_node);
1015             list_push_front(&dst->dst_arcs, &arc->dst_node);
1016             arc->src = src;
1017             arc->dst = dst;
1018         }
1019
1020         return !ovsdb_idl_row_is_orphan(dst) ? dst : NULL;
1021     }
1022 }
1023
1024 const struct ovsdb_idl_row *
1025 ovsdb_idl_get_row_for_uuid(const struct ovsdb_idl *idl,
1026                            const struct ovsdb_idl_table_class *tc,
1027                            const struct uuid *uuid)
1028 {
1029     return ovsdb_idl_get_row(ovsdb_idl_table_from_class(idl, tc), uuid);
1030 }
1031
1032 static struct ovsdb_idl_row *
1033 next_real_row(struct ovsdb_idl_table *table, struct hmap_node *node)
1034 {
1035     for (; node; node = hmap_next(&table->rows, node)) {
1036         struct ovsdb_idl_row *row;
1037
1038         row = CONTAINER_OF(node, struct ovsdb_idl_row, hmap_node);
1039         if (ovsdb_idl_row_exists(row)) {
1040             return row;
1041         }
1042     }
1043     return NULL;
1044 }
1045
1046 const struct ovsdb_idl_row *
1047 ovsdb_idl_first_row(const struct ovsdb_idl *idl,
1048                     const struct ovsdb_idl_table_class *table_class)
1049 {
1050     struct ovsdb_idl_table *table
1051         = ovsdb_idl_table_from_class(idl, table_class);
1052     return next_real_row(table, hmap_first(&table->rows));
1053 }
1054
1055 const struct ovsdb_idl_row *
1056 ovsdb_idl_next_row(const struct ovsdb_idl_row *row)
1057 {
1058     struct ovsdb_idl_table *table = row->table;
1059
1060     return next_real_row(table, hmap_next(&table->rows, &row->hmap_node));
1061 }
1062
1063 /* Reads and returns the value of 'column' within 'row'.  If an ongoing
1064  * transaction has changed 'column''s value, the modified value is returned.
1065  *
1066  * The caller must not modify or free the returned value.
1067  *
1068  * Various kinds of changes can invalidate the returned value: writing to the
1069  * same 'column' in 'row' (e.g. with ovsdb_idl_txn_write()), deleting 'row'
1070  * (e.g. with ovsdb_idl_txn_delete()), or completing an ongoing transaction
1071  * (e.g. with ovsdb_idl_txn_commit() or ovsdb_idl_txn_abort()).  If the
1072  * returned value is needed for a long time, it is best to make a copy of it
1073  * with ovsdb_datum_clone(). */
1074 const struct ovsdb_datum *
1075 ovsdb_idl_read(const struct ovsdb_idl_row *row,
1076                const struct ovsdb_idl_column *column)
1077 {
1078     const struct ovsdb_idl_table_class *class = row->table->class;
1079     size_t column_idx = column - class->columns;
1080
1081     assert(row->new != NULL);
1082     assert(column_idx < class->n_columns);
1083
1084     if (row->written && bitmap_is_set(row->written, column_idx)) {
1085         return &row->new[column_idx];
1086     } else if (row->old) {
1087         return &row->old[column_idx];
1088     } else {
1089         return ovsdb_datum_default(&column->type);
1090     }
1091 }
1092
1093 /* Same as ovsdb_idl_read(), except that it also asserts that 'column' has key
1094  * type 'key_type' and value type 'value_type'.  (Scalar and set types will
1095  * have a value type of OVSDB_TYPE_VOID.)
1096  *
1097  * This is useful in code that "knows" that a particular column has a given
1098  * type, so that it will abort if someone changes the column's type without
1099  * updating the code that uses it. */
1100 const struct ovsdb_datum *
1101 ovsdb_idl_get(const struct ovsdb_idl_row *row,
1102               const struct ovsdb_idl_column *column,
1103               enum ovsdb_atomic_type key_type OVS_UNUSED,
1104               enum ovsdb_atomic_type value_type OVS_UNUSED)
1105 {
1106     assert(column->type.key.type == key_type);
1107     assert(column->type.value.type == value_type);
1108
1109     return ovsdb_idl_read(row, column);
1110 }
1111
1112 /* Returns false if 'row' was obtained from the IDL, true if it was initialized
1113  * to all-zero-bits by some other entity.  If 'row' was set up some other way
1114  * then the return value is indeterminate. */
1115 bool
1116 ovsdb_idl_row_is_synthetic(const struct ovsdb_idl_row *row)
1117 {
1118     return row->table == NULL;
1119 }
1120 \f
1121 /* Transactions. */
1122
1123 static void ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
1124                                    enum ovsdb_idl_txn_status);
1125
1126 const char *
1127 ovsdb_idl_txn_status_to_string(enum ovsdb_idl_txn_status status)
1128 {
1129     switch (status) {
1130     case TXN_UNCHANGED:
1131         return "unchanged";
1132     case TXN_INCOMPLETE:
1133         return "incomplete";
1134     case TXN_ABORTED:
1135         return "aborted";
1136     case TXN_SUCCESS:
1137         return "success";
1138     case TXN_TRY_AGAIN:
1139         return "try again";
1140     case TXN_ERROR:
1141         return "error";
1142     }
1143     return "<unknown>";
1144 }
1145
1146 struct ovsdb_idl_txn *
1147 ovsdb_idl_txn_create(struct ovsdb_idl *idl)
1148 {
1149     struct ovsdb_idl_txn *txn;
1150
1151     assert(!idl->txn);
1152     idl->txn = txn = xmalloc(sizeof *txn);
1153     txn->request_id = NULL;
1154     txn->idl = idl;
1155     hmap_init(&txn->txn_rows);
1156     txn->status = TXN_INCOMPLETE;
1157     txn->error = NULL;
1158     txn->dry_run = false;
1159     ds_init(&txn->comment);
1160
1161     txn->inc_table = NULL;
1162     txn->inc_column = NULL;
1163     txn->inc_where = NULL;
1164
1165     hmap_init(&txn->inserted_rows);
1166
1167     return txn;
1168 }
1169
1170 /* Appends 's', which is treated as a printf()-type format string, to the
1171  * comments that will be passed to the OVSDB server when 'txn' is committed.
1172  * (The comment will be committed to the OVSDB log, which "ovsdb-tool
1173  * show-log" can print in a relatively human-readable form.) */
1174 void
1175 ovsdb_idl_txn_add_comment(struct ovsdb_idl_txn *txn, const char *s, ...)
1176 {
1177     va_list args;
1178
1179     if (txn->comment.length) {
1180         ds_put_char(&txn->comment, '\n');
1181     }
1182
1183     va_start(args, s);
1184     ds_put_format_valist(&txn->comment, s, args);
1185     va_end(args);
1186 }
1187
1188 void
1189 ovsdb_idl_txn_set_dry_run(struct ovsdb_idl_txn *txn)
1190 {
1191     txn->dry_run = true;
1192 }
1193
1194 void
1195 ovsdb_idl_txn_increment(struct ovsdb_idl_txn *txn, const char *table,
1196                         const char *column, const struct json *where)
1197 {
1198     assert(!txn->inc_table);
1199     txn->inc_table = xstrdup(table);
1200     txn->inc_column = xstrdup(column);
1201     txn->inc_where = where ? json_clone(where) : json_array_create_empty();
1202 }
1203
1204 void
1205 ovsdb_idl_txn_destroy(struct ovsdb_idl_txn *txn)
1206 {
1207     struct ovsdb_idl_txn_insert *insert, *next;
1208
1209     json_destroy(txn->request_id);
1210     if (txn->status == TXN_INCOMPLETE) {
1211         hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
1212     }
1213     ovsdb_idl_txn_abort(txn);
1214     ds_destroy(&txn->comment);
1215     free(txn->error);
1216     free(txn->inc_table);
1217     free(txn->inc_column);
1218     json_destroy(txn->inc_where);
1219     HMAP_FOR_EACH_SAFE (insert, next, hmap_node, &txn->inserted_rows) {
1220         free(insert);
1221     }
1222     hmap_destroy(&txn->inserted_rows);
1223     free(txn);
1224 }
1225
1226 void
1227 ovsdb_idl_txn_wait(const struct ovsdb_idl_txn *txn)
1228 {
1229     if (txn->status != TXN_INCOMPLETE) {
1230         poll_immediate_wake();
1231     }
1232 }
1233
1234 static struct json *
1235 where_uuid_equals(const struct uuid *uuid)
1236 {
1237     return
1238         json_array_create_1(
1239             json_array_create_3(
1240                 json_string_create("_uuid"),
1241                 json_string_create("=="),
1242                 json_array_create_2(
1243                     json_string_create("uuid"),
1244                     json_string_create_nocopy(
1245                         xasprintf(UUID_FMT, UUID_ARGS(uuid))))));
1246 }
1247
1248 static char *
1249 uuid_name_from_uuid(const struct uuid *uuid)
1250 {
1251     char *name;
1252     char *p;
1253
1254     name = xasprintf("row"UUID_FMT, UUID_ARGS(uuid));
1255     for (p = name; *p != '\0'; p++) {
1256         if (*p == '-') {
1257             *p = '_';
1258         }
1259     }
1260
1261     return name;
1262 }
1263
1264 static const struct ovsdb_idl_row *
1265 ovsdb_idl_txn_get_row(const struct ovsdb_idl_txn *txn, const struct uuid *uuid)
1266 {
1267     const struct ovsdb_idl_row *row;
1268
1269     HMAP_FOR_EACH_WITH_HASH (row, txn_node, uuid_hash(uuid), &txn->txn_rows) {
1270         if (uuid_equals(&row->uuid, uuid)) {
1271             return row;
1272         }
1273     }
1274     return NULL;
1275 }
1276
1277 /* XXX there must be a cleaner way to do this */
1278 static struct json *
1279 substitute_uuids(struct json *json, const struct ovsdb_idl_txn *txn)
1280 {
1281     if (json->type == JSON_ARRAY) {
1282         struct uuid uuid;
1283         size_t i;
1284
1285         if (json->u.array.n == 2
1286             && json->u.array.elems[0]->type == JSON_STRING
1287             && json->u.array.elems[1]->type == JSON_STRING
1288             && !strcmp(json->u.array.elems[0]->u.string, "uuid")
1289             && uuid_from_string(&uuid, json->u.array.elems[1]->u.string)) {
1290             const struct ovsdb_idl_row *row;
1291
1292             row = ovsdb_idl_txn_get_row(txn, &uuid);
1293             if (row && !row->old && row->new) {
1294                 json_destroy(json);
1295
1296                 return json_array_create_2(
1297                     json_string_create("named-uuid"),
1298                     json_string_create_nocopy(uuid_name_from_uuid(&uuid)));
1299             }
1300         }
1301
1302         for (i = 0; i < json->u.array.n; i++) {
1303             json->u.array.elems[i] = substitute_uuids(json->u.array.elems[i],
1304                                                       txn);
1305         }
1306     } else if (json->type == JSON_OBJECT) {
1307         struct shash_node *node;
1308
1309         SHASH_FOR_EACH (node, json_object(json)) {
1310             node->data = substitute_uuids(node->data, txn);
1311         }
1312     }
1313     return json;
1314 }
1315
1316 static void
1317 ovsdb_idl_txn_disassemble(struct ovsdb_idl_txn *txn)
1318 {
1319     struct ovsdb_idl_row *row, *next;
1320
1321     /* This must happen early.  Otherwise, ovsdb_idl_row_parse() will call an
1322      * ovsdb_idl_column's 'parse' function, which will call
1323      * ovsdb_idl_get_row_arc(), which will seen that the IDL is in a
1324      * transaction and fail to update the graph.  */
1325     txn->idl->txn = NULL;
1326
1327     HMAP_FOR_EACH_SAFE (row, next, txn_node, &txn->txn_rows) {
1328         if (row->old) {
1329             if (row->written) {
1330                 ovsdb_idl_row_unparse(row);
1331                 ovsdb_idl_row_clear_arcs(row, false);
1332                 ovsdb_idl_row_parse(row);
1333             }
1334         } else {
1335             ovsdb_idl_row_unparse(row);
1336         }
1337         ovsdb_idl_row_clear_new(row);
1338
1339         free(row->prereqs);
1340         row->prereqs = NULL;
1341
1342         free(row->written);
1343         row->written = NULL;
1344
1345         hmap_remove(&txn->txn_rows, &row->txn_node);
1346         hmap_node_nullify(&row->txn_node);
1347         if (!row->old) {
1348             hmap_remove(&row->table->rows, &row->hmap_node);
1349             free(row);
1350         }
1351     }
1352     hmap_destroy(&txn->txn_rows);
1353     hmap_init(&txn->txn_rows);
1354 }
1355
1356 enum ovsdb_idl_txn_status
1357 ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
1358 {
1359     struct ovsdb_idl_row *row;
1360     struct json *operations;
1361     bool any_updates;
1362
1363     if (txn != txn->idl->txn) {
1364         return txn->status;
1365     }
1366
1367     operations = json_array_create_1(
1368         json_string_create(txn->idl->class->database));
1369
1370     /* Add prerequisites and declarations of new rows. */
1371     HMAP_FOR_EACH (row, txn_node, &txn->txn_rows) {
1372         /* XXX check that deleted rows exist even if no prereqs? */
1373         if (row->prereqs) {
1374             const struct ovsdb_idl_table_class *class = row->table->class;
1375             size_t n_columns = class->n_columns;
1376             struct json *op, *columns, *row_json;
1377             size_t idx;
1378
1379             op = json_object_create();
1380             json_array_add(operations, op);
1381             json_object_put_string(op, "op", "wait");
1382             json_object_put_string(op, "table", class->name);
1383             json_object_put(op, "timeout", json_integer_create(0));
1384             json_object_put(op, "where", where_uuid_equals(&row->uuid));
1385             json_object_put_string(op, "until", "==");
1386             columns = json_array_create_empty();
1387             json_object_put(op, "columns", columns);
1388             row_json = json_object_create();
1389             json_object_put(op, "rows", json_array_create_1(row_json));
1390
1391             BITMAP_FOR_EACH_1 (idx, n_columns, row->prereqs) {
1392                 const struct ovsdb_idl_column *column = &class->columns[idx];
1393                 json_array_add(columns, json_string_create(column->name));
1394                 json_object_put(row_json, column->name,
1395                                 ovsdb_datum_to_json(&row->old[idx],
1396                                                     &column->type));
1397             }
1398         }
1399     }
1400
1401     /* Add updates. */
1402     any_updates = false;
1403     HMAP_FOR_EACH (row, txn_node, &txn->txn_rows) {
1404         const struct ovsdb_idl_table_class *class = row->table->class;
1405
1406         if (row->old == row->new) {
1407             continue;
1408         } else if (!row->new) {
1409             if (class->is_root) {
1410                 struct json *op = json_object_create();
1411                 json_object_put_string(op, "op", "delete");
1412                 json_object_put_string(op, "table", class->name);
1413                 json_object_put(op, "where", where_uuid_equals(&row->uuid));
1414                 json_array_add(operations, op);
1415                 any_updates = true;
1416             } else {
1417                 /* Let ovsdb-server decide whether to really delete it. */
1418             }
1419         } else {
1420             struct json *row_json;
1421             struct json *op;
1422             size_t idx;
1423
1424             op = json_object_create();
1425             json_object_put_string(op, "op", row->old ? "update" : "insert");
1426             json_object_put_string(op, "table", class->name);
1427             if (row->old) {
1428                 json_object_put(op, "where", where_uuid_equals(&row->uuid));
1429             } else {
1430                 struct ovsdb_idl_txn_insert *insert;
1431
1432                 json_object_put(op, "uuid-name",
1433                                 json_string_create_nocopy(
1434                                     uuid_name_from_uuid(&row->uuid)));
1435
1436                 insert = xmalloc(sizeof *insert);
1437                 insert->dummy = row->uuid;
1438                 insert->op_index = operations->u.array.n - 1;
1439                 uuid_zero(&insert->real);
1440                 hmap_insert(&txn->inserted_rows, &insert->hmap_node,
1441                             uuid_hash(&insert->dummy));
1442             }
1443             row_json = json_object_create();
1444             json_object_put(op, "row", row_json);
1445
1446             if (row->written) {
1447                 BITMAP_FOR_EACH_1 (idx, class->n_columns, row->written) {
1448                     const struct ovsdb_idl_column *column =
1449                                                         &class->columns[idx];
1450
1451                     if (row->old
1452                         || !ovsdb_datum_is_default(&row->new[idx],
1453                                                   &column->type)) {
1454                         json_object_put(row_json, column->name,
1455                                         substitute_uuids(
1456                                             ovsdb_datum_to_json(&row->new[idx],
1457                                                                 &column->type),
1458                                             txn));
1459                     }
1460                 }
1461             }
1462
1463             if (!row->old || !shash_is_empty(json_object(row_json))) {
1464                 json_array_add(operations, op);
1465                 any_updates = true;
1466             } else {
1467                 json_destroy(op);
1468             }
1469         }
1470     }
1471
1472     /* Add increment. */
1473     if (txn->inc_table && any_updates) {
1474         struct json *op;
1475
1476         txn->inc_index = operations->u.array.n - 1;
1477
1478         op = json_object_create();
1479         json_object_put_string(op, "op", "mutate");
1480         json_object_put_string(op, "table", txn->inc_table);
1481         json_object_put(op, "where",
1482                         substitute_uuids(json_clone(txn->inc_where), txn));
1483         json_object_put(op, "mutations",
1484                         json_array_create_1(
1485                             json_array_create_3(
1486                                 json_string_create(txn->inc_column),
1487                                 json_string_create("+="),
1488                                 json_integer_create(1))));
1489         json_array_add(operations, op);
1490
1491         op = json_object_create();
1492         json_object_put_string(op, "op", "select");
1493         json_object_put_string(op, "table", txn->inc_table);
1494         json_object_put(op, "where",
1495                         substitute_uuids(json_clone(txn->inc_where), txn));
1496         json_object_put(op, "columns",
1497                         json_array_create_1(json_string_create(
1498                                                 txn->inc_column)));
1499         json_array_add(operations, op);
1500     }
1501
1502     if (txn->comment.length) {
1503         struct json *op = json_object_create();
1504         json_object_put_string(op, "op", "comment");
1505         json_object_put_string(op, "comment", ds_cstr(&txn->comment));
1506         json_array_add(operations, op);
1507     }
1508
1509     if (txn->dry_run) {
1510         struct json *op = json_object_create();
1511         json_object_put_string(op, "op", "abort");
1512         json_array_add(operations, op);
1513     }
1514
1515     if (!any_updates) {
1516         txn->status = TXN_UNCHANGED;
1517         json_destroy(operations);
1518     } else if (!jsonrpc_session_send(
1519                    txn->idl->session,
1520                    jsonrpc_create_request(
1521                        "transact", operations, &txn->request_id))) {
1522         hmap_insert(&txn->idl->outstanding_txns, &txn->hmap_node,
1523                     json_hash(txn->request_id, 0));
1524     } else {
1525         txn->status = TXN_TRY_AGAIN;
1526     }
1527
1528     ovsdb_idl_txn_disassemble(txn);
1529     return txn->status;
1530 }
1531
1532 /* Attempts to commit 'txn', blocking until the commit either succeeds or
1533  * fails.  Returns the final commit status, which may be any TXN_* value other
1534  * than TXN_INCOMPLETE. */
1535 enum ovsdb_idl_txn_status
1536 ovsdb_idl_txn_commit_block(struct ovsdb_idl_txn *txn)
1537 {
1538     enum ovsdb_idl_txn_status status;
1539
1540     fatal_signal_run();
1541     while ((status = ovsdb_idl_txn_commit(txn)) == TXN_INCOMPLETE) {
1542         ovsdb_idl_run(txn->idl);
1543         ovsdb_idl_wait(txn->idl);
1544         ovsdb_idl_txn_wait(txn);
1545         poll_block();
1546     }
1547     return status;
1548 }
1549
1550 int64_t
1551 ovsdb_idl_txn_get_increment_new_value(const struct ovsdb_idl_txn *txn)
1552 {
1553     assert(txn->status == TXN_SUCCESS);
1554     return txn->inc_new_value;
1555 }
1556
1557 void
1558 ovsdb_idl_txn_abort(struct ovsdb_idl_txn *txn)
1559 {
1560     ovsdb_idl_txn_disassemble(txn);
1561     if (txn->status == TXN_INCOMPLETE) {
1562         txn->status = TXN_ABORTED;
1563     }
1564 }
1565
1566 const char *
1567 ovsdb_idl_txn_get_error(const struct ovsdb_idl_txn *txn)
1568 {
1569     if (txn->status != TXN_ERROR) {
1570         return ovsdb_idl_txn_status_to_string(txn->status);
1571     } else if (txn->error) {
1572         return txn->error;
1573     } else {
1574         return "no error details available";
1575     }
1576 }
1577
1578 static void
1579 ovsdb_idl_txn_set_error_json(struct ovsdb_idl_txn *txn,
1580                              const struct json *json)
1581 {
1582     if (txn->error == NULL) {
1583         txn->error = json_to_string(json, JSSF_SORT);
1584     }
1585 }
1586
1587 /* For transaction 'txn' that completed successfully, finds and returns the
1588  * permanent UUID that the database assigned to a newly inserted row, given the
1589  * 'uuid' that ovsdb_idl_txn_insert() assigned locally to that row.
1590  *
1591  * Returns NULL if 'uuid' is not a UUID assigned by ovsdb_idl_txn_insert() or
1592  * if it was assigned by that function and then deleted by
1593  * ovsdb_idl_txn_delete() within the same transaction.  (Rows that are inserted
1594  * and then deleted within a single transaction are never sent to the database
1595  * server, so it never assigns them a permanent UUID.) */
1596 const struct uuid *
1597 ovsdb_idl_txn_get_insert_uuid(const struct ovsdb_idl_txn *txn,
1598                               const struct uuid *uuid)
1599 {
1600     const struct ovsdb_idl_txn_insert *insert;
1601
1602     assert(txn->status == TXN_SUCCESS || txn->status == TXN_UNCHANGED);
1603     HMAP_FOR_EACH_IN_BUCKET (insert, hmap_node,
1604                              uuid_hash(uuid), &txn->inserted_rows) {
1605         if (uuid_equals(uuid, &insert->dummy)) {
1606             return &insert->real;
1607         }
1608     }
1609     return NULL;
1610 }
1611
1612 static void
1613 ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
1614                        enum ovsdb_idl_txn_status status)
1615 {
1616     txn->status = status;
1617     hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
1618 }
1619
1620 /* Writes 'datum' to the specified 'column' in 'row_'.  Updates both 'row_'
1621  * itself and the structs derived from it (e.g. the "struct ovsrec_*", for
1622  * ovs-vswitchd).
1623  *
1624  * 'datum' must have the correct type for its column.  The IDL does not check
1625  * that it meets schema constraints, but ovsdb-server will do so at commit time
1626  * so it had better be correct.
1627  *
1628  * A transaction must be in progress.  Replication of 'column' must not have
1629  * been disabled (by calling ovsdb_idl_omit()).
1630  *
1631  * Usually this function is used indirectly through one of the "set" functions
1632  * generated by ovsdb-idlc. */
1633 void
1634 ovsdb_idl_txn_write(const struct ovsdb_idl_row *row_,
1635                     const struct ovsdb_idl_column *column,
1636                     struct ovsdb_datum *datum)
1637 {
1638     struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
1639     const struct ovsdb_idl_table_class *class = row->table->class;
1640     size_t column_idx = column - class->columns;
1641
1642     assert(row->new != NULL);
1643     assert(column_idx < class->n_columns);
1644     assert(row->old == NULL ||
1645            row->table->modes[column_idx] & OVSDB_IDL_MONITOR);
1646
1647     /* If this is a write-only column and the datum being written is the same
1648      * as the one already there, just skip the update entirely.  This is worth
1649      * optimizing because we have a lot of columns that get periodically
1650      * refreshed into the database but don't actually change that often.
1651      *
1652      * We don't do this for read/write columns because that would break
1653      * atomicity of transactions--some other client might have written a
1654      * different value in that column since we read it. */
1655     if (row->table->modes[column_idx] == OVSDB_IDL_MONITOR
1656         && ovsdb_datum_equals(ovsdb_idl_read(row, column),
1657                               datum, &column->type)) {
1658         ovsdb_datum_destroy(datum, &column->type);
1659         return;
1660     }
1661
1662     if (hmap_node_is_null(&row->txn_node)) {
1663         hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
1664                     uuid_hash(&row->uuid));
1665     }
1666     if (row->old == row->new) {
1667         row->new = xmalloc(class->n_columns * sizeof *row->new);
1668     }
1669     if (!row->written) {
1670         row->written = bitmap_allocate(class->n_columns);
1671     }
1672     if (bitmap_is_set(row->written, column_idx)) {
1673         ovsdb_datum_destroy(&row->new[column_idx], &column->type);
1674     } else {
1675         bitmap_set1(row->written, column_idx);
1676     }
1677     row->new[column_idx] = *datum;
1678     (column->unparse)(row);
1679     (column->parse)(row, &row->new[column_idx]);
1680 }
1681
1682 /* Causes the original contents of 'column' in 'row_' to be verified as a
1683  * prerequisite to completing the transaction.  That is, if 'column' in 'row_'
1684  * changed (or if 'row_' was deleted) between the time that the IDL originally
1685  * read its contents and the time that the transaction commits, then the
1686  * transaction aborts and ovsdb_idl_txn_commit() returns TXN_TRY_AGAIN.
1687  *
1688  * The intention is that, to ensure that no transaction commits based on dirty
1689  * reads, an application should call ovsdb_idl_txn_verify() on each data item
1690  * read as part of a read-modify-write operation.
1691  *
1692  * In some cases ovsdb_idl_txn_verify() reduces to a no-op, because the current
1693  * value of 'column' is already known:
1694  *
1695  *   - If 'row_' is a row created by the current transaction (returned by
1696  *     ovsdb_idl_txn_insert()).
1697  *
1698  *   - If 'column' has already been modified (with ovsdb_idl_txn_write())
1699  *     within the current transaction.
1700  *
1701  * Because of the latter property, always call ovsdb_idl_txn_verify() *before*
1702  * ovsdb_idl_txn_write() for a given read-modify-write.
1703  *
1704  * A transaction must be in progress.
1705  *
1706  * Usually this function is used indirectly through one of the "verify"
1707  * functions generated by ovsdb-idlc. */
1708 void
1709 ovsdb_idl_txn_verify(const struct ovsdb_idl_row *row_,
1710                      const struct ovsdb_idl_column *column)
1711 {
1712     struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
1713     const struct ovsdb_idl_table_class *class = row->table->class;
1714     size_t column_idx = column - class->columns;
1715
1716     assert(row->new != NULL);
1717     assert(row->old == NULL ||
1718            row->table->modes[column_idx] & OVSDB_IDL_MONITOR);
1719     if (!row->old
1720         || (row->written && bitmap_is_set(row->written, column_idx))) {
1721         return;
1722     }
1723
1724     if (hmap_node_is_null(&row->txn_node)) {
1725         hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
1726                     uuid_hash(&row->uuid));
1727     }
1728     if (!row->prereqs) {
1729         row->prereqs = bitmap_allocate(class->n_columns);
1730     }
1731     bitmap_set1(row->prereqs, column_idx);
1732 }
1733
1734 /* Deletes 'row_' from its table.  May free 'row_', so it must not be
1735  * accessed afterward.
1736  *
1737  * A transaction must be in progress.
1738  *
1739  * Usually this function is used indirectly through one of the "delete"
1740  * functions generated by ovsdb-idlc. */
1741 void
1742 ovsdb_idl_txn_delete(const struct ovsdb_idl_row *row_)
1743 {
1744     struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
1745
1746     assert(row->new != NULL);
1747     if (!row->old) {
1748         ovsdb_idl_row_unparse(row);
1749         ovsdb_idl_row_clear_new(row);
1750         assert(!row->prereqs);
1751         hmap_remove(&row->table->rows, &row->hmap_node);
1752         hmap_remove(&row->table->idl->txn->txn_rows, &row->txn_node);
1753         free(row);
1754         return;
1755     }
1756     if (hmap_node_is_null(&row->txn_node)) {
1757         hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
1758                     uuid_hash(&row->uuid));
1759     }
1760     ovsdb_idl_row_clear_new(row);
1761     row->new = NULL;
1762 }
1763
1764 /* Inserts and returns a new row in the table with the specified 'class' in the
1765  * database with open transaction 'txn'.
1766  *
1767  * The new row is assigned a provisional UUID.  If 'uuid' is null then one is
1768  * randomly generated; otherwise 'uuid' should specify a randomly generated
1769  * UUID not otherwise in use.  ovsdb-server will assign a different UUID when
1770  * 'txn' is committed, but the IDL will replace any uses of the provisional
1771  * UUID in the data to be to be committed by the UUID assigned by
1772  * ovsdb-server.
1773  *
1774  * Usually this function is used indirectly through one of the "insert"
1775  * functions generated by ovsdb-idlc. */
1776 const struct ovsdb_idl_row *
1777 ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
1778                      const struct ovsdb_idl_table_class *class,
1779                      const struct uuid *uuid)
1780 {
1781     struct ovsdb_idl_row *row = ovsdb_idl_row_create__(class);
1782
1783     if (uuid) {
1784         assert(!ovsdb_idl_txn_get_row(txn, uuid));
1785         row->uuid = *uuid;
1786     } else {
1787         uuid_generate(&row->uuid);
1788     }
1789
1790     row->table = ovsdb_idl_table_from_class(txn->idl, class);
1791     row->new = xmalloc(class->n_columns * sizeof *row->new);
1792     hmap_insert(&row->table->rows, &row->hmap_node, uuid_hash(&row->uuid));
1793     hmap_insert(&txn->txn_rows, &row->txn_node, uuid_hash(&row->uuid));
1794     return row;
1795 }
1796
1797 static void
1798 ovsdb_idl_txn_abort_all(struct ovsdb_idl *idl)
1799 {
1800     struct ovsdb_idl_txn *txn;
1801
1802     HMAP_FOR_EACH (txn, hmap_node, &idl->outstanding_txns) {
1803         ovsdb_idl_txn_complete(txn, TXN_TRY_AGAIN);
1804     }
1805 }
1806
1807 static struct ovsdb_idl_txn *
1808 ovsdb_idl_txn_find(struct ovsdb_idl *idl, const struct json *id)
1809 {
1810     struct ovsdb_idl_txn *txn;
1811
1812     HMAP_FOR_EACH_WITH_HASH (txn, hmap_node,
1813                              json_hash(id, 0), &idl->outstanding_txns) {
1814         if (json_equal(id, txn->request_id)) {
1815             return txn;
1816         }
1817     }
1818     return NULL;
1819 }
1820
1821 static bool
1822 check_json_type(const struct json *json, enum json_type type, const char *name)
1823 {
1824     if (!json) {
1825         VLOG_WARN_RL(&syntax_rl, "%s is missing", name);
1826         return false;
1827     } else if (json->type != type) {
1828         VLOG_WARN_RL(&syntax_rl, "%s is %s instead of %s",
1829                      name, json_type_to_string(json->type),
1830                      json_type_to_string(type));
1831         return false;
1832     } else {
1833         return true;
1834     }
1835 }
1836
1837 static bool
1838 ovsdb_idl_txn_process_inc_reply(struct ovsdb_idl_txn *txn,
1839                                 const struct json_array *results)
1840 {
1841     struct json *count, *rows, *row, *column;
1842     struct shash *mutate, *select;
1843
1844     if (txn->inc_index + 2 > results->n) {
1845         VLOG_WARN_RL(&syntax_rl, "reply does not contain enough operations "
1846                      "for increment (has %zu, needs %u)",
1847                      results->n, txn->inc_index + 2);
1848         return false;
1849     }
1850
1851     /* We know that this is a JSON object because the loop in
1852      * ovsdb_idl_txn_process_reply() checked. */
1853     mutate = json_object(results->elems[txn->inc_index]);
1854     count = shash_find_data(mutate, "count");
1855     if (!check_json_type(count, JSON_INTEGER, "\"mutate\" reply \"count\"")) {
1856         return false;
1857     }
1858     if (count->u.integer != 1) {
1859         VLOG_WARN_RL(&syntax_rl,
1860                      "\"mutate\" reply \"count\" is %lld instead of 1",
1861                      count->u.integer);
1862         return false;
1863     }
1864
1865     select = json_object(results->elems[txn->inc_index + 1]);
1866     rows = shash_find_data(select, "rows");
1867     if (!check_json_type(rows, JSON_ARRAY, "\"select\" reply \"rows\"")) {
1868         return false;
1869     }
1870     if (rows->u.array.n != 1) {
1871         VLOG_WARN_RL(&syntax_rl, "\"select\" reply \"rows\" has %zu elements "
1872                      "instead of 1",
1873                      rows->u.array.n);
1874         return false;
1875     }
1876     row = rows->u.array.elems[0];
1877     if (!check_json_type(row, JSON_OBJECT, "\"select\" reply row")) {
1878         return false;
1879     }
1880     column = shash_find_data(json_object(row), txn->inc_column);
1881     if (!check_json_type(column, JSON_INTEGER,
1882                          "\"select\" reply inc column")) {
1883         return false;
1884     }
1885     txn->inc_new_value = column->u.integer;
1886     return true;
1887 }
1888
1889 static bool
1890 ovsdb_idl_txn_process_insert_reply(struct ovsdb_idl_txn_insert *insert,
1891                                    const struct json_array *results)
1892 {
1893     static const struct ovsdb_base_type uuid_type = OVSDB_BASE_UUID_INIT;
1894     struct ovsdb_error *error;
1895     struct json *json_uuid;
1896     union ovsdb_atom uuid;
1897     struct shash *reply;
1898
1899     if (insert->op_index >= results->n) {
1900         VLOG_WARN_RL(&syntax_rl, "reply does not contain enough operations "
1901                      "for insert (has %zu, needs %u)",
1902                      results->n, insert->op_index);
1903         return false;
1904     }
1905
1906     /* We know that this is a JSON object because the loop in
1907      * ovsdb_idl_txn_process_reply() checked. */
1908     reply = json_object(results->elems[insert->op_index]);
1909     json_uuid = shash_find_data(reply, "uuid");
1910     if (!check_json_type(json_uuid, JSON_ARRAY, "\"insert\" reply \"uuid\"")) {
1911         return false;
1912     }
1913
1914     error = ovsdb_atom_from_json(&uuid, &uuid_type, json_uuid, NULL);
1915     if (error) {
1916         char *s = ovsdb_error_to_string(error);
1917         VLOG_WARN_RL(&syntax_rl, "\"insert\" reply \"uuid\" is not a JSON "
1918                      "UUID: %s", s);
1919         free(s);
1920         return false;
1921     }
1922
1923     insert->real = uuid.uuid;
1924
1925     return true;
1926 }
1927
1928 static bool
1929 ovsdb_idl_txn_process_reply(struct ovsdb_idl *idl,
1930                             const struct jsonrpc_msg *msg)
1931 {
1932     struct ovsdb_idl_txn *txn;
1933     enum ovsdb_idl_txn_status status;
1934
1935     txn = ovsdb_idl_txn_find(idl, msg->id);
1936     if (!txn) {
1937         return false;
1938     }
1939
1940     if (msg->type == JSONRPC_ERROR) {
1941         status = TXN_ERROR;
1942     } else if (msg->result->type != JSON_ARRAY) {
1943         VLOG_WARN_RL(&syntax_rl, "reply to \"transact\" is not JSON array");
1944         status = TXN_ERROR;
1945     } else {
1946         struct json_array *ops = &msg->result->u.array;
1947         int hard_errors = 0;
1948         int soft_errors = 0;
1949         size_t i;
1950
1951         for (i = 0; i < ops->n; i++) {
1952             struct json *op = ops->elems[i];
1953
1954             if (op->type == JSON_NULL) {
1955                 /* This isn't an error in itself but indicates that some prior
1956                  * operation failed, so make sure that we know about it. */
1957                 soft_errors++;
1958             } else if (op->type == JSON_OBJECT) {
1959                 struct json *error;
1960
1961                 error = shash_find_data(json_object(op), "error");
1962                 if (error) {
1963                     if (error->type == JSON_STRING) {
1964                         if (!strcmp(error->u.string, "timed out")) {
1965                             soft_errors++;
1966                         } else if (strcmp(error->u.string, "aborted")) {
1967                             hard_errors++;
1968                             ovsdb_idl_txn_set_error_json(txn, op);
1969                         }
1970                     } else {
1971                         hard_errors++;
1972                         ovsdb_idl_txn_set_error_json(txn, op);
1973                         VLOG_WARN_RL(&syntax_rl,
1974                                      "\"error\" in reply is not JSON string");
1975                     }
1976                 }
1977             } else {
1978                 hard_errors++;
1979                 ovsdb_idl_txn_set_error_json(txn, op);
1980                 VLOG_WARN_RL(&syntax_rl,
1981                              "operation reply is not JSON null or object");
1982             }
1983         }
1984
1985         if (!soft_errors && !hard_errors) {
1986             struct ovsdb_idl_txn_insert *insert;
1987
1988             if (txn->inc_table && !ovsdb_idl_txn_process_inc_reply(txn, ops)) {
1989                 hard_errors++;
1990             }
1991
1992             HMAP_FOR_EACH (insert, hmap_node, &txn->inserted_rows) {
1993                 if (!ovsdb_idl_txn_process_insert_reply(insert, ops)) {
1994                     hard_errors++;
1995                 }
1996             }
1997         }
1998
1999         status = (hard_errors ? TXN_ERROR
2000                   : soft_errors ? TXN_TRY_AGAIN
2001                   : TXN_SUCCESS);
2002     }
2003
2004     ovsdb_idl_txn_complete(txn, status);
2005     return true;
2006 }
2007
2008 struct ovsdb_idl_txn *
2009 ovsdb_idl_txn_get(const struct ovsdb_idl_row *row)
2010 {
2011     struct ovsdb_idl_txn *txn = row->table->idl->txn;
2012     assert(txn != NULL);
2013     return txn;
2014 }
2015
2016 struct ovsdb_idl *
2017 ovsdb_idl_txn_get_idl (struct ovsdb_idl_txn *txn)
2018 {
2019     return txn->idl;
2020 }
2021