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