lib/ovsdb-idl.c

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