lib/ovsdb-idl.c

   1 /* Copyright (c) 2009, 2010 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
  38 #define THIS_MODULE VLM_ovsdb_idl
  39 #include "vlog.h"
  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;
  70     struct json *monitor_request_id;
  71     unsigned int last_monitor_request_seqno;
  72     unsigned int change_seqno;
  73
  74     /* Transaction support. */
  75     struct ovsdb_idl_txn *txn;
  76     struct hmap outstanding_txns;
  77 };
  78
  79 struct ovsdb_idl_txn {
  80     struct hmap_node hmap_node;
  81     struct json *request_id;
  82     struct ovsdb_idl *idl;
  83     struct hmap txn_rows;
  84     enum ovsdb_idl_txn_status status;
  85     char *error;
  86     bool dry_run;
  87     struct ds comment;
  88
  89     /* Increments. */
  90     char *inc_table;
  91     char *inc_column;
  92     struct json *inc_where;
  93     unsigned int inc_index;
  94     int64_t inc_new_value;
  95
  96     /* Inserted rows. */
  97     struct hmap inserted_rows;
  98 };
  99
 100 struct ovsdb_idl_txn_insert {
 101     struct hmap_node hmap_node; /* In struct ovsdb_idl_txn's inserted_rows. */
 102     struct uuid dummy;          /* Dummy UUID used locally. */
 103     int op_index;               /* Index into transaction's operation array. */
 104     struct uuid real;           /* Real UUID used by database server. */
 105 };
 106
 107 static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);
 108 static struct vlog_rate_limit semantic_rl = VLOG_RATE_LIMIT_INIT(1, 5);
 109
 110 static void ovsdb_idl_clear(struct ovsdb_idl *);
 111 static void ovsdb_idl_send_monitor_request(struct ovsdb_idl *);
 112 static void ovsdb_idl_parse_update(struct ovsdb_idl *, const struct json *);
 113 static struct ovsdb_error *ovsdb_idl_parse_update__(struct ovsdb_idl *,
 114                                                     const struct json *);
 115 static void ovsdb_idl_process_update(struct ovsdb_idl_table *,
 116                                      const struct uuid *,
 117                                      const struct json *old,
 118                                      const struct json *new);
 119 static void ovsdb_idl_insert_row(struct ovsdb_idl_row *, const struct json *);
 120 static void ovsdb_idl_delete_row(struct ovsdb_idl_row *);
 121 static void ovsdb_idl_modify_row(struct ovsdb_idl_row *, const struct json *);
 122
 123 static bool ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *);
 124 static struct ovsdb_idl_row *ovsdb_idl_row_create__(
 125     const struct ovsdb_idl_table_class *);
 126 static struct ovsdb_idl_row *ovsdb_idl_row_create(struct ovsdb_idl_table *,
 127                                                   const struct uuid *);
 128 static void ovsdb_idl_row_destroy(struct ovsdb_idl_row *);
 129
 130 static void ovsdb_idl_row_parse(struct ovsdb_idl_row *);
 131 static void ovsdb_idl_row_unparse(struct ovsdb_idl_row *);
 132 static void ovsdb_idl_row_clear_old(struct ovsdb_idl_row *);
 133 static void ovsdb_idl_row_clear_new(struct ovsdb_idl_row *);
 134
 135 static void ovsdb_idl_txn_abort_all(struct ovsdb_idl *);
 136 static bool ovsdb_idl_txn_process_reply(struct ovsdb_idl *,
 137                                         const struct jsonrpc_msg *msg);
 138
 139 struct ovsdb_idl *
 140 ovsdb_idl_create(const char *remote, const struct ovsdb_idl_class *class)
 141 {
 142     struct ovsdb_idl *idl;
 143     size_t i;
 144
 145     idl = xzalloc(sizeof *idl);
 146     idl->class = class;
 147     idl->session = jsonrpc_session_open(remote);
 148     shash_init(&idl->table_by_name);
 149     idl->tables = xmalloc(class->n_tables * sizeof *idl->tables);
 150     for (i = 0; i < class->n_tables; i++) {
 151         const struct ovsdb_idl_table_class *tc = &class->tables[i];
 152         struct ovsdb_idl_table *table = &idl->tables[i];
 153         size_t j;
 154
 155         assert(!shash_find(&idl->table_by_name, tc->name));
 156         shash_add(&idl->table_by_name, tc->name, table);
 157         table->class = tc;
 158         shash_init(&table->columns);
 159         for (j = 0; j < tc->n_columns; j++) {
 160             const struct ovsdb_idl_column *column = &tc->columns[j];
 161
 162             assert(!shash_find(&table->columns, column->name));
 163             shash_add(&table->columns, column->name, column);
 164         }
 165         hmap_init(&table->rows);
 166         table->idl = idl;
 167     }
 168     idl->last_monitor_request_seqno = UINT_MAX;
 169     hmap_init(&idl->outstanding_txns);
 170
 171     return idl;
 172 }
 173
 174 void
 175 ovsdb_idl_destroy(struct ovsdb_idl *idl)
 176 {
 177     if (idl) {
 178         size_t i;
 179
 180         assert(!idl->txn);
 181         ovsdb_idl_clear(idl);
 182         jsonrpc_session_close(idl->session);
 183
 184         for (i = 0; i < idl->class->n_tables; i++) {
 185             struct ovsdb_idl_table *table = &idl->tables[i];
 186             shash_destroy(&table->columns);
 187             hmap_destroy(&table->rows);
 188         }
 189         shash_destroy(&idl->table_by_name);
 190         free(idl->tables);
 191         json_destroy(idl->monitor_request_id);
 192         free(idl);
 193     }
 194 }
 195
 196 static void
 197 ovsdb_idl_clear(struct ovsdb_idl *idl)
 198 {
 199     bool changed = false;
 200     size_t i;
 201
 202     for (i = 0; i < idl->class->n_tables; i++) {
 203         struct ovsdb_idl_table *table = &idl->tables[i];
 204         struct ovsdb_idl_row *row, *next_row;
 205
 206         if (hmap_is_empty(&table->rows)) {
 207             continue;
 208         }
 209
 210         changed = true;
 211         HMAP_FOR_EACH_SAFE (row, next_row, struct ovsdb_idl_row, hmap_node,
 212                             &table->rows) {
 213             struct ovsdb_idl_arc *arc, *next_arc;
 214
 215             if (!ovsdb_idl_row_is_orphan(row)) {
 216                 ovsdb_idl_row_unparse(row);
 217             }
 218             LIST_FOR_EACH_SAFE (arc, next_arc, struct ovsdb_idl_arc, src_node,
 219                                 &row->src_arcs) {
 220                 free(arc);
 221             }
 222             /* No need to do anything with dst_arcs: some node has those arcs
 223              * as forward arcs and will destroy them itself. */
 224
 225             ovsdb_idl_row_destroy(row);
 226         }
 227     }
 228
 229     if (changed) {
 230         idl->change_seqno++;
 231     }
 232 }
 233
 234 bool
 235 ovsdb_idl_run(struct ovsdb_idl *idl)
 236 {
 237     unsigned int initial_change_seqno = idl->change_seqno;
 238     int i;
 239
 240     assert(!idl->txn);
 241     jsonrpc_session_run(idl->session);
 242     for (i = 0; jsonrpc_session_is_connected(idl->session) && i < 50; i++) {
 243         struct jsonrpc_msg *msg, *reply;
 244         unsigned int seqno;
 245
 246         seqno = jsonrpc_session_get_seqno(idl->session);
 247         if (idl->last_monitor_request_seqno != seqno) {
 248             idl->last_monitor_request_seqno = seqno;
 249             ovsdb_idl_txn_abort_all(idl);
 250             ovsdb_idl_send_monitor_request(idl);
 251             break;
 252         }
 253
 254         msg = jsonrpc_session_recv(idl->session);
 255         if (!msg) {
 256             break;
 257         }
 258
 259         reply = NULL;
 260         if (msg->type == JSONRPC_NOTIFY
 261                    && !strcmp(msg->method, "update")
 262                    && msg->params->type == JSON_ARRAY
 263                    && msg->params->u.array.n == 2
 264                    && msg->params->u.array.elems[0]->type == JSON_NULL) {
 265             ovsdb_idl_parse_update(idl, msg->params->u.array.elems[1]);
 266         } else if (msg->type == JSONRPC_REPLY
 267                    && idl->monitor_request_id
 268                    && json_equal(idl->monitor_request_id, msg->id)) {
 269             json_destroy(idl->monitor_request_id);
 270             idl->monitor_request_id = NULL;
 271             ovsdb_idl_clear(idl);
 272             ovsdb_idl_parse_update(idl, msg->result);
 273         } else if (msg->type == JSONRPC_REPLY
 274                    && msg->id && msg->id->type == JSON_STRING
 275                    && !strcmp(msg->id->u.string, "echo")) {
 276             /* It's a reply to our echo request.  Ignore it. */
 277         } else if ((msg->type == JSONRPC_ERROR
 278                     || msg->type == JSONRPC_REPLY)
 279                    && ovsdb_idl_txn_process_reply(idl, msg)) {
 280             /* ovsdb_idl_txn_process_reply() did everything needful. */
 281         } else {
 282             /* This can happen if ovsdb_idl_txn_destroy() is called to destroy
 283              * a transaction before we receive the reply, so keep the log level
 284              * low. */
 285             VLOG_DBG("%s: received unexpected %s message",
 286                      jsonrpc_session_get_name(idl->session),
 287                      jsonrpc_msg_type_to_string(msg->type));
 288         }
 289         if (reply) {
 290             jsonrpc_session_send(idl->session, reply);
 291         }
 292         jsonrpc_msg_destroy(msg);
 293     }
 294
 295     return initial_change_seqno != idl->change_seqno;
 296 }
 297
 298 void
 299 ovsdb_idl_wait(struct ovsdb_idl *idl)
 300 {
 301     jsonrpc_session_wait(idl->session);
 302     jsonrpc_session_recv_wait(idl->session);
 303 }
 304
 305 unsigned int
 306 ovsdb_idl_get_seqno(const struct ovsdb_idl *idl)
 307 {
 308     return idl->change_seqno;
 309 }
 310
 311 bool
 312 ovsdb_idl_has_ever_connected(const struct ovsdb_idl *idl)
 313 {
 314     return ovsdb_idl_get_seqno(idl) != 0;
 315 }
 316
 317 void
 318 ovsdb_idl_force_reconnect(struct ovsdb_idl *idl)
 319 {
 320     jsonrpc_session_force_reconnect(idl->session);
 321 }
 322 \f
 323 static void
 324 ovsdb_idl_send_monitor_request(struct ovsdb_idl *idl)
 325 {
 326     struct json *monitor_requests;
 327     struct jsonrpc_msg *msg;
 328     size_t i;
 329
 330     monitor_requests = json_object_create();
 331     for (i = 0; i < idl->class->n_tables; i++) {
 332         const struct ovsdb_idl_table *table = &idl->tables[i];
 333         const struct ovsdb_idl_table_class *tc = table->class;
 334         struct json *monitor_request, *columns;
 335         size_t i;
 336
 337         monitor_request = json_object_create();
 338         columns = json_array_create_empty();
 339         for (i = 0; i < tc->n_columns; i++) {
 340             const struct ovsdb_idl_column *column = &tc->columns[i];
 341             json_array_add(columns, json_string_create(column->name));
 342         }
 343         json_object_put(monitor_request, "columns", columns);
 344         json_object_put(monitor_requests, tc->name, monitor_request);
 345     }
 346
 347     json_destroy(idl->monitor_request_id);
 348     msg = jsonrpc_create_request(
 349         "monitor",
 350         json_array_create_3(json_string_create(idl->class->database),
 351                             json_null_create(), monitor_requests),
 352         &idl->monitor_request_id);
 353     jsonrpc_session_send(idl->session, msg);
 354 }
 355
 356 static void
 357 ovsdb_idl_parse_update(struct ovsdb_idl *idl, const struct json *table_updates)
 358 {
 359     struct ovsdb_error *error;
 360
 361     idl->change_seqno++;
 362
 363     error = ovsdb_idl_parse_update__(idl, table_updates);
 364     if (error) {
 365         if (!VLOG_DROP_WARN(&syntax_rl)) {
 366             char *s = ovsdb_error_to_string(error);
 367             VLOG_WARN_RL(&syntax_rl, "%s", s);
 368             free(s);
 369         }
 370         ovsdb_error_destroy(error);
 371     }
 372 }
 373
 374 static struct ovsdb_error *
 375 ovsdb_idl_parse_update__(struct ovsdb_idl *idl,
 376                          const struct json *table_updates)
 377 {
 378     const struct shash_node *tables_node;
 379
 380     if (table_updates->type != JSON_OBJECT) {
 381         return ovsdb_syntax_error(table_updates, NULL,
 382                                   "<table-updates> is not an object");
 383     }
 384     SHASH_FOR_EACH (tables_node, json_object(table_updates)) {
 385         const struct json *table_update = tables_node->data;
 386         const struct shash_node *table_node;
 387         struct ovsdb_idl_table *table;
 388
 389         table = shash_find_data(&idl->table_by_name, tables_node->name);
 390         if (!table) {
 391             return ovsdb_syntax_error(
 392                 table_updates, NULL,
 393                 "<table-updates> includes unknown table \"%s\"",
 394                 tables_node->name);
 395         }
 396
 397         if (table_update->type != JSON_OBJECT) {
 398             return ovsdb_syntax_error(table_update, NULL,
 399                                       "<table-update> for table \"%s\" is "
 400                                       "not an object", table->class->name);
 401         }
 402         SHASH_FOR_EACH (table_node, json_object(table_update)) {
 403             const struct json *row_update = table_node->data;
 404             const struct json *old_json, *new_json;
 405             struct uuid uuid;
 406
 407             if (!uuid_from_string(&uuid, table_node->name)) {
 408                 return ovsdb_syntax_error(table_update, NULL,
 409                                           "<table-update> for table \"%s\" "
 410                                           "contains bad UUID "
 411                                           "\"%s\" as member name",
 412                                           table->class->name,
 413                                           table_node->name);
 414             }
 415             if (row_update->type != JSON_OBJECT) {
 416                 return ovsdb_syntax_error(row_update, NULL,
 417                                           "<table-update> for table \"%s\" "
 418                                           "contains <row-update> for %s that "
 419                                           "is not an object",
 420                                           table->class->name,
 421                                           table_node->name);
 422             }
 423
 424             old_json = shash_find_data(json_object(row_update), "old");
 425             new_json = shash_find_data(json_object(row_update), "new");
 426             if (old_json && old_json->type != JSON_OBJECT) {
 427                 return ovsdb_syntax_error(old_json, NULL,
 428                                           "\"old\" <row> is not object");
 429             } else if (new_json && new_json->type != JSON_OBJECT) {
 430                 return ovsdb_syntax_error(new_json, NULL,
 431                                           "\"new\" <row> is not object");
 432             } else if ((old_json != NULL) + (new_json != NULL)
 433                        != shash_count(json_object(row_update))) {
 434                 return ovsdb_syntax_error(row_update, NULL,
 435                                           "<row-update> contains unexpected "
 436                                           "member");
 437             } else if (!old_json && !new_json) {
 438                 return ovsdb_syntax_error(row_update, NULL,
 439                                           "<row-update> missing \"old\" "
 440                                           "and \"new\" members");
 441             }
 442
 443             ovsdb_idl_process_update(table, &uuid, old_json, new_json);
 444         }
 445     }
 446
 447     return NULL;
 448 }
 449
 450 static struct ovsdb_idl_row *
 451 ovsdb_idl_get_row(struct ovsdb_idl_table *table, const struct uuid *uuid)
 452 {
 453     struct ovsdb_idl_row *row;
 454
 455     HMAP_FOR_EACH_WITH_HASH (row, struct ovsdb_idl_row, hmap_node,
 456                              uuid_hash(uuid), &table->rows) {
 457         if (uuid_equals(&row->uuid, uuid)) {
 458             return row;
 459         }
 460     }
 461     return NULL;
 462 }
 463
 464 static void
 465 ovsdb_idl_process_update(struct ovsdb_idl_table *table,
 466                          const struct uuid *uuid, const struct json *old,
 467                          const struct json *new)
 468 {
 469     struct ovsdb_idl_row *row;
 470
 471     row = ovsdb_idl_get_row(table, uuid);
 472     if (!new) {
 473         /* Delete row. */
 474         if (row && !ovsdb_idl_row_is_orphan(row)) {
 475             /* XXX perhaps we should check the 'old' values? */
 476             ovsdb_idl_delete_row(row);
 477         } else {
 478             VLOG_WARN_RL(&semantic_rl, "cannot delete missing row "UUID_FMT" "
 479                          "from table %s",
 480                          UUID_ARGS(uuid), table->class->name);
 481         }
 482     } else if (!old) {
 483         /* Insert row. */
 484         if (!row) {
 485             ovsdb_idl_insert_row(ovsdb_idl_row_create(table, uuid), new);
 486         } else if (ovsdb_idl_row_is_orphan(row)) {
 487             ovsdb_idl_insert_row(row, new);
 488         } else {
 489             VLOG_WARN_RL(&semantic_rl, "cannot add existing row "UUID_FMT" to "
 490                          "table %s", UUID_ARGS(uuid), table->class->name);
 491             ovsdb_idl_modify_row(row, new);
 492         }
 493     } else {
 494         /* Modify row. */
 495         if (row) {
 496             /* XXX perhaps we should check the 'old' values? */
 497             if (!ovsdb_idl_row_is_orphan(row)) {
 498                 ovsdb_idl_modify_row(row, new);
 499             } else {
 500                 VLOG_WARN_RL(&semantic_rl, "cannot modify missing but "
 501                              "referenced row "UUID_FMT" in table %s",
 502                              UUID_ARGS(uuid), table->class->name);
 503                 ovsdb_idl_insert_row(row, new);
 504             }
 505         } else {
 506             VLOG_WARN_RL(&semantic_rl, "cannot modify missing row "UUID_FMT" "
 507                          "in table %s", UUID_ARGS(uuid), table->class->name);
 508             ovsdb_idl_insert_row(ovsdb_idl_row_create(table, uuid), new);
 509         }
 510     }
 511 }
 512
 513 static void
 514 ovsdb_idl_row_update(struct ovsdb_idl_row *row, const struct json *row_json)
 515 {
 516     struct ovsdb_idl_table *table = row->table;
 517     struct shash_node *node;
 518
 519     SHASH_FOR_EACH (node, json_object(row_json)) {
 520         const char *column_name = node->name;
 521         const struct ovsdb_idl_column *column;
 522         struct ovsdb_datum datum;
 523         struct ovsdb_error *error;
 524
 525         column = shash_find_data(&table->columns, column_name);
 526         if (!column) {
 527             VLOG_WARN_RL(&syntax_rl, "unknown column %s updating row "UUID_FMT,
 528                          column_name, UUID_ARGS(&row->uuid));
 529             continue;
 530         }
 531
 532         error = ovsdb_datum_from_json(&datum, &column->type, node->data, NULL);
 533         if (!error) {
 534             ovsdb_datum_swap(&row->old[column - table->class->columns],
 535                              &datum);
 536             ovsdb_datum_destroy(&datum, &column->type);
 537         } else {
 538             char *s = ovsdb_error_to_string(error);
 539             VLOG_WARN_RL(&syntax_rl, "error parsing column %s in row "UUID_FMT
 540                          " in table %s: %s", column_name,
 541                          UUID_ARGS(&row->uuid), table->class->name, s);
 542             free(s);
 543             ovsdb_error_destroy(error);
 544         }
 545     }
 546 }
 547
 548 /* When a row A refers to row B through a column with a "refTable" constraint,
 549  * but row B does not exist, row B is called an "orphan row".  Orphan rows
 550  * should not persist, because the database enforces referential integrity, but
 551  * they can appear transiently as changes from the database are received (the
 552  * database doesn't try to topologically sort them and circular references mean
 553  * it isn't always possible anyhow).
 554  *
 555  * This function returns true if 'row' is an orphan row, otherwise false.
 556  */
 557 static bool
 558 ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *row)
 559 {
 560     return !row->old && !row->new;
 561 }
 562
 563 /* Returns true if 'row' is conceptually part of the database as modified by
 564  * the current transaction (if any), false otherwise.
 565  *
 566  * This function will return true if 'row' is not an orphan (see the comment on
 567  * ovsdb_idl_row_is_orphan()) and:
 568  *
 569  *   - 'row' exists in the database and has not been deleted within the
 570  *     current transaction (if any).
 571  *
 572  *   - 'row' was inserted within the current transaction and has not been
 573  *     deleted.  (In the latter case you should not have passed 'row' in at
 574  *     all, because ovsdb_idl_txn_delete() freed it.)
 575  *
 576  * This function will return false if 'row' is an orphan or if 'row' was
 577  * deleted within the current transaction.
 578  */
 579 static bool
 580 ovsdb_idl_row_exists(const struct ovsdb_idl_row *row)
 581 {
 582     return row->new != NULL;
 583 }
 584
 585 static void
 586 ovsdb_idl_row_parse(struct ovsdb_idl_row *row)
 587 {
 588     const struct ovsdb_idl_table_class *class = row->table->class;
 589     size_t i;
 590
 591     for (i = 0; i < class->n_columns; i++) {
 592         const struct ovsdb_idl_column *c = &class->columns[i];
 593         (c->parse)(row, &row->old[i]);
 594     }
 595 }
 596
 597 static void
 598 ovsdb_idl_row_unparse(struct ovsdb_idl_row *row)
 599 {
 600     const struct ovsdb_idl_table_class *class = row->table->class;
 601     size_t i;
 602
 603     for (i = 0; i < class->n_columns; i++) {
 604         const struct ovsdb_idl_column *c = &class->columns[i];
 605         (c->unparse)(row);
 606     }
 607 }
 608
 609 static void
 610 ovsdb_idl_row_clear_old(struct ovsdb_idl_row *row)
 611 {
 612     assert(row->old == row->new);
 613     if (!ovsdb_idl_row_is_orphan(row)) {
 614         const struct ovsdb_idl_table_class *class = row->table->class;
 615         size_t i;
 616
 617         for (i = 0; i < class->n_columns; i++) {
 618             ovsdb_datum_destroy(&row->old[i], &class->columns[i].type);
 619         }
 620         free(row->old);
 621         row->old = row->new = NULL;
 622     }
 623 }
 624
 625 static void
 626 ovsdb_idl_row_clear_new(struct ovsdb_idl_row *row)
 627 {
 628     if (row->old != row->new) {
 629         if (row->new) {
 630             const struct ovsdb_idl_table_class *class = row->table->class;
 631             size_t i;
 632
 633             BITMAP_FOR_EACH_1 (i, class->n_columns, row->written) {
 634                 ovsdb_datum_destroy(&row->new[i], &class->columns[i].type);
 635             }
 636             free(row->new);
 637             free(row->written);
 638             row->written = NULL;
 639         }
 640         row->new = row->old;
 641     }
 642 }
 643
 644 static void
 645 ovsdb_idl_row_clear_arcs(struct ovsdb_idl_row *row, bool destroy_dsts)
 646 {
 647     struct ovsdb_idl_arc *arc, *next;
 648
 649     /* Delete all forward arcs.  If 'destroy_dsts', destroy any orphaned rows
 650      * that this causes to be unreferenced. */
 651     LIST_FOR_EACH_SAFE (arc, next, struct ovsdb_idl_arc, src_node,
 652                         &row->src_arcs) {
 653         list_remove(&arc->dst_node);
 654         if (destroy_dsts
 655             && ovsdb_idl_row_is_orphan(arc->dst)
 656             && list_is_empty(&arc->dst->dst_arcs)) {
 657             ovsdb_idl_row_destroy(arc->dst);
 658         }
 659         free(arc);
 660     }
 661     list_init(&row->src_arcs);
 662 }
 663
 664 /* Force nodes that reference 'row' to reparse. */
 665 static void
 666 ovsdb_idl_row_reparse_backrefs(struct ovsdb_idl_row *row)
 667 {
 668     struct ovsdb_idl_arc *arc, *next;
 669
 670     /* This is trickier than it looks.  ovsdb_idl_row_clear_arcs() will destroy
 671      * 'arc', so we need to use the "safe" variant of list traversal.  However,
 672      * calling an ovsdb_idl_column's 'parse' function will add an arc
 673      * equivalent to 'arc' to row->arcs.  That could be a problem for
 674      * traversal, but it adds it at the beginning of the list to prevent us
 675      * from stumbling upon it again.
 676      *
 677      * (If duplicate arcs were possible then we would need to make sure that
 678      * 'next' didn't also point into 'arc''s destination, but we forbid
 679      * duplicate arcs.) */
 680     LIST_FOR_EACH_SAFE (arc, next, struct ovsdb_idl_arc, dst_node,
 681                         &row->dst_arcs) {
 682         struct ovsdb_idl_row *ref = arc->src;
 683
 684         ovsdb_idl_row_unparse(ref);
 685         ovsdb_idl_row_clear_arcs(ref, false);
 686         ovsdb_idl_row_parse(ref);
 687     }
 688 }
 689
 690 static struct ovsdb_idl_row *
 691 ovsdb_idl_row_create__(const struct ovsdb_idl_table_class *class)
 692 {
 693     struct ovsdb_idl_row *row = xzalloc(class->allocation_size);
 694     list_init(&row->src_arcs);
 695     list_init(&row->dst_arcs);
 696     hmap_node_nullify(&row->txn_node);
 697     return row;
 698 }
 699
 700 static struct ovsdb_idl_row *
 701 ovsdb_idl_row_create(struct ovsdb_idl_table *table, const struct uuid *uuid)
 702 {
 703     struct ovsdb_idl_row *row = ovsdb_idl_row_create__(table->class);
 704     hmap_insert(&table->rows, &row->hmap_node, uuid_hash(uuid));
 705     row->uuid = *uuid;
 706     row->table = table;
 707     return row;
 708 }
 709
 710 static void
 711 ovsdb_idl_row_destroy(struct ovsdb_idl_row *row)
 712 {
 713     if (row) {
 714         ovsdb_idl_row_clear_old(row);
 715         hmap_remove(&row->table->rows, &row->hmap_node);
 716         free(row);
 717     }
 718 }
 719
 720 static void
 721 ovsdb_idl_insert_row(struct ovsdb_idl_row *row, const struct json *row_json)
 722 {
 723     const struct ovsdb_idl_table_class *class = row->table->class;
 724     size_t i;
 725
 726     assert(!row->old && !row->new);
 727     row->old = row->new = xmalloc(class->n_columns * sizeof *row->old);
 728     for (i = 0; i < class->n_columns; i++) {
 729         ovsdb_datum_init_default(&row->old[i], &class->columns[i].type);
 730     }
 731     ovsdb_idl_row_update(row, row_json);
 732     ovsdb_idl_row_parse(row);
 733
 734     ovsdb_idl_row_reparse_backrefs(row);
 735 }
 736
 737 static void
 738 ovsdb_idl_delete_row(struct ovsdb_idl_row *row)
 739 {
 740     ovsdb_idl_row_unparse(row);
 741     ovsdb_idl_row_clear_arcs(row, true);
 742     ovsdb_idl_row_clear_old(row);
 743     if (list_is_empty(&row->dst_arcs)) {
 744         ovsdb_idl_row_destroy(row);
 745     } else {
 746         ovsdb_idl_row_reparse_backrefs(row);
 747     }
 748 }
 749
 750 static void
 751 ovsdb_idl_modify_row(struct ovsdb_idl_row *row, const struct json *row_json)
 752 {
 753     ovsdb_idl_row_unparse(row);
 754     ovsdb_idl_row_clear_arcs(row, true);
 755     ovsdb_idl_row_update(row, row_json);
 756     ovsdb_idl_row_parse(row);
 757 }
 758
 759 static bool
 760 may_add_arc(const struct ovsdb_idl_row *src, const struct ovsdb_idl_row *dst)
 761 {
 762     const struct ovsdb_idl_arc *arc;
 763
 764     /* No self-arcs. */
 765     if (src == dst) {
 766         return false;
 767     }
 768
 769     /* No duplicate arcs.
 770      *
 771      * We only need to test whether the first arc in dst->dst_arcs originates
 772      * at 'src', since we add all of the arcs from a given source in a clump
 773      * (in a single call to ovsdb_idl_row_parse()) and new arcs are always
 774      * added at the front of the dst_arcs list. */
 775     if (list_is_empty(&dst->dst_arcs)) {
 776         return true;
 777     }
 778     arc = CONTAINER_OF(dst->dst_arcs.next, struct ovsdb_idl_arc, dst_node);
 779     return arc->src != src;
 780 }
 781
 782 static struct ovsdb_idl_table *
 783 ovsdb_idl_table_from_class(const struct ovsdb_idl *idl,
 784                            const struct ovsdb_idl_table_class *table_class)
 785 {
 786     return &idl->tables[table_class - idl->class->tables];
 787 }
 788
 789 struct ovsdb_idl_row *
 790 ovsdb_idl_get_row_arc(struct ovsdb_idl_row *src,
 791                       struct ovsdb_idl_table_class *dst_table_class,
 792                       const struct uuid *dst_uuid)
 793 {
 794     struct ovsdb_idl *idl = src->table->idl;
 795     struct ovsdb_idl_table *dst_table;
 796     struct ovsdb_idl_arc *arc;
 797     struct ovsdb_idl_row *dst;
 798
 799     dst_table = ovsdb_idl_table_from_class(idl, dst_table_class);
 800     dst = ovsdb_idl_get_row(dst_table, dst_uuid);
 801     if (idl->txn) {
 802         /* We're being called from ovsdb_idl_txn_write().  We must not update
 803          * any arcs, because the transaction will be backed out at commit or
 804          * abort time and we don't want our graph screwed up.
 805          *
 806          * Just return the destination row, if there is one and it has not been
 807          * deleted. */
 808         if (dst && (hmap_node_is_null(&dst->txn_node) || dst->new)) {
 809             return dst;
 810         }
 811         return NULL;
 812     } else {
 813         /* We're being called from some other context.  Update the graph. */
 814         if (!dst) {
 815             dst = ovsdb_idl_row_create(dst_table, dst_uuid);
 816         }
 817
 818         /* Add a new arc, if it wouldn't be a self-arc or a duplicate arc. */
 819         if (may_add_arc(src, dst)) {
 820             /* The arc *must* be added at the front of the dst_arcs list.  See
 821              * ovsdb_idl_row_reparse_backrefs() for details. */
 822             arc = xmalloc(sizeof *arc);
 823             list_push_front(&src->src_arcs, &arc->src_node);
 824             list_push_front(&dst->dst_arcs, &arc->dst_node);
 825             arc->src = src;
 826             arc->dst = dst;
 827         }
 828
 829         return !ovsdb_idl_row_is_orphan(dst) ? dst : NULL;
 830     }
 831 }
 832
 833 const struct ovsdb_idl_row *
 834 ovsdb_idl_get_row_for_uuid(const struct ovsdb_idl *idl,
 835                            const struct ovsdb_idl_table_class *tc,
 836                            const struct uuid *uuid)
 837 {
 838     return ovsdb_idl_get_row(ovsdb_idl_table_from_class(idl, tc), uuid);
 839 }
 840
 841 static struct ovsdb_idl_row *
 842 next_real_row(struct ovsdb_idl_table *table, struct hmap_node *node)
 843 {
 844     for (; node; node = hmap_next(&table->rows, node)) {
 845         struct ovsdb_idl_row *row;
 846
 847         row = CONTAINER_OF(node, struct ovsdb_idl_row, hmap_node);
 848         if (ovsdb_idl_row_exists(row)) {
 849             return row;
 850         }
 851     }
 852     return NULL;
 853 }
 854
 855 const struct ovsdb_idl_row *
 856 ovsdb_idl_first_row(const struct ovsdb_idl *idl,
 857                     const struct ovsdb_idl_table_class *table_class)
 858 {
 859     struct ovsdb_idl_table *table
 860         = ovsdb_idl_table_from_class(idl, table_class);
 861     return next_real_row(table, hmap_first(&table->rows));
 862 }
 863
 864 const struct ovsdb_idl_row *
 865 ovsdb_idl_next_row(const struct ovsdb_idl_row *row)
 866 {
 867     struct ovsdb_idl_table *table = row->table;
 868
 869     return next_real_row(table, hmap_next(&table->rows, &row->hmap_node));
 870 }
 871 \f
 872 /* Transactions. */
 873
 874 static void ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
 875                                    enum ovsdb_idl_txn_status);
 876
 877 const char *
 878 ovsdb_idl_txn_status_to_string(enum ovsdb_idl_txn_status status)
 879 {
 880     switch (status) {
 881     case TXN_UNCHANGED:
 882         return "unchanged";
 883     case TXN_INCOMPLETE:
 884         return "incomplete";
 885     case TXN_ABORTED:
 886         return "aborted";
 887     case TXN_SUCCESS:
 888         return "success";
 889     case TXN_TRY_AGAIN:
 890         return "try again";
 891     case TXN_ERROR:
 892         return "error";
 893     }
 894     return "<unknown>";
 895 }
 896
 897 struct ovsdb_idl_txn *
 898 ovsdb_idl_txn_create(struct ovsdb_idl *idl)
 899 {
 900     struct ovsdb_idl_txn *txn;
 901
 902     assert(!idl->txn);
 903     idl->txn = txn = xmalloc(sizeof *txn);
 904     txn->request_id = NULL;
 905     txn->idl = idl;
 906     hmap_init(&txn->txn_rows);
 907     txn->status = TXN_INCOMPLETE;
 908     txn->error = NULL;
 909     txn->dry_run = false;
 910     ds_init(&txn->comment);
 911
 912     txn->inc_table = NULL;
 913     txn->inc_column = NULL;
 914     txn->inc_where = NULL;
 915
 916     hmap_init(&txn->inserted_rows);
 917
 918     return txn;
 919 }
 920
 921 /* Appends 's', which is treated as a printf()-type format string, to the
 922  * comments that will be passed to the OVSDB server when 'txn' is committed.
 923  * (The comment will be committed to the OVSDB log, which "ovsdb-tool
 924  * show-log" can print in a relatively human-readable form.) */
 925 void
 926 ovsdb_idl_txn_add_comment(struct ovsdb_idl_txn *txn, const char *s, ...)
 927 {
 928     va_list args;
 929
 930     if (txn->comment.length) {
 931         ds_put_char(&txn->comment, '\n');
 932     }
 933
 934     va_start(args, s);
 935     ds_put_format_valist(&txn->comment, s, args);
 936     va_end(args);
 937 }
 938
 939 void
 940 ovsdb_idl_txn_set_dry_run(struct ovsdb_idl_txn *txn)
 941 {
 942     txn->dry_run = true;
 943 }
 944
 945 void
 946 ovsdb_idl_txn_increment(struct ovsdb_idl_txn *txn, const char *table,
 947                         const char *column, const struct json *where)
 948 {
 949     assert(!txn->inc_table);
 950     txn->inc_table = xstrdup(table);
 951     txn->inc_column = xstrdup(column);
 952     txn->inc_where = where ? json_clone(where) : json_array_create_empty();
 953 }
 954
 955 void
 956 ovsdb_idl_txn_destroy(struct ovsdb_idl_txn *txn)
 957 {
 958     struct ovsdb_idl_txn_insert *insert, *next;
 959
 960     json_destroy(txn->request_id);
 961     if (txn->status == TXN_INCOMPLETE) {
 962         hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
 963     }
 964     ovsdb_idl_txn_abort(txn);
 965     ds_destroy(&txn->comment);
 966     free(txn->error);
 967     free(txn->inc_table);
 968     free(txn->inc_column);
 969     json_destroy(txn->inc_where);
 970     HMAP_FOR_EACH_SAFE (insert, next, struct ovsdb_idl_txn_insert, hmap_node,
 971                         &txn->inserted_rows) {
 972         free(insert);
 973     }
 974     hmap_destroy(&txn->inserted_rows);
 975     free(txn);
 976 }
 977
 978 void
 979 ovsdb_idl_txn_wait(const struct ovsdb_idl_txn *txn)
 980 {
 981     if (txn->status != TXN_INCOMPLETE) {
 982         poll_immediate_wake();
 983     }
 984 }
 985
 986 static struct json *
 987 where_uuid_equals(const struct uuid *uuid)
 988 {
 989     return
 990         json_array_create_1(
 991             json_array_create_3(
 992                 json_string_create("_uuid"),
 993                 json_string_create("=="),
 994                 json_array_create_2(
 995                     json_string_create("uuid"),
 996                     json_string_create_nocopy(
 997                         xasprintf(UUID_FMT, UUID_ARGS(uuid))))));
 998 }
 999
1000 static char *
1001 uuid_name_from_uuid(const struct uuid *uuid)
1002 {
1003     char *name;
1004     char *p;
1005
1006     name = xasprintf("row"UUID_FMT, UUID_ARGS(uuid));
1007     for (p = name; *p != '\0'; p++) {
1008         if (*p == '-') {
1009             *p = '_';
1010         }
1011     }
1012
1013     return name;
1014 }
1015
1016 static const struct ovsdb_idl_row *
1017 ovsdb_idl_txn_get_row(const struct ovsdb_idl_txn *txn, const struct uuid *uuid)
1018 {
1019     const struct ovsdb_idl_row *row;
1020
1021     HMAP_FOR_EACH_WITH_HASH (row, struct ovsdb_idl_row, txn_node,
1022                              uuid_hash(uuid), &txn->txn_rows) {
1023         if (uuid_equals(&row->uuid, uuid)) {
1024             return row;
1025         }
1026     }
1027     return NULL;
1028 }
1029
1030 /* XXX there must be a cleaner way to do this */
1031 static struct json *
1032 substitute_uuids(struct json *json, const struct ovsdb_idl_txn *txn)
1033 {
1034     if (json->type == JSON_ARRAY) {
1035         struct uuid uuid;
1036         size_t i;
1037
1038         if (json->u.array.n == 2
1039             && json->u.array.elems[0]->type == JSON_STRING
1040             && json->u.array.elems[1]->type == JSON_STRING
1041             && !strcmp(json->u.array.elems[0]->u.string, "uuid")
1042             && uuid_from_string(&uuid, json->u.array.elems[1]->u.string)) {
1043             const struct ovsdb_idl_row *row;
1044
1045             row = ovsdb_idl_txn_get_row(txn, &uuid);
1046             if (row && !row->old && row->new) {
1047                 json_destroy(json);
1048
1049                 return json_array_create_2(
1050                     json_string_create("named-uuid"),
1051                     json_string_create_nocopy(uuid_name_from_uuid(&uuid)));
1052             }
1053         }
1054
1055         for (i = 0; i < json->u.array.n; i++) {
1056             json->u.array.elems[i] = substitute_uuids(json->u.array.elems[i],
1057                                                       txn);
1058         }
1059     } else if (json->type == JSON_OBJECT) {
1060         struct shash_node *node;
1061
1062         SHASH_FOR_EACH (node, json_object(json)) {
1063             node->data = substitute_uuids(node->data, txn);
1064         }
1065     }
1066     return json;
1067 }
1068
1069 static void
1070 ovsdb_idl_txn_disassemble(struct ovsdb_idl_txn *txn)
1071 {
1072     struct ovsdb_idl_row *row, *next;
1073
1074     /* This must happen early.  Otherwise, ovsdb_idl_row_parse() will call an
1075      * ovsdb_idl_column's 'parse' function, which will call
1076      * ovsdb_idl_get_row_arc(), which will seen that the IDL is in a
1077      * transaction and fail to update the graph.  */
1078     txn->idl->txn = NULL;
1079
1080     HMAP_FOR_EACH_SAFE (row, next, struct ovsdb_idl_row, txn_node,
1081                         &txn->txn_rows) {
1082         if (row->old) {
1083             if (row->written) {
1084                 ovsdb_idl_row_unparse(row);
1085                 ovsdb_idl_row_clear_arcs(row, false);
1086                 ovsdb_idl_row_parse(row);
1087             }
1088         } else {
1089             ovsdb_idl_row_unparse(row);
1090         }
1091         ovsdb_idl_row_clear_new(row);
1092
1093         free(row->prereqs);
1094         row->prereqs = NULL;
1095
1096         free(row->written);
1097         row->written = NULL;
1098
1099         hmap_remove(&txn->txn_rows, &row->txn_node);
1100         hmap_node_nullify(&row->txn_node);
1101         if (!row->old) {
1102             hmap_remove(&row->table->rows, &row->hmap_node);
1103             free(row);
1104         }
1105     }
1106     hmap_destroy(&txn->txn_rows);
1107     hmap_init(&txn->txn_rows);
1108 }
1109
1110 enum ovsdb_idl_txn_status
1111 ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
1112 {
1113     struct ovsdb_idl_row *row;
1114     struct json *operations;
1115     bool any_updates;
1116
1117     if (txn != txn->idl->txn) {
1118         return txn->status;
1119     }
1120
1121     operations = json_array_create_1(
1122         json_string_create(txn->idl->class->database));
1123
1124     /* Add prerequisites and declarations of new rows. */
1125     HMAP_FOR_EACH (row, struct ovsdb_idl_row, txn_node, &txn->txn_rows) {
1126         /* XXX check that deleted rows exist even if no prereqs? */
1127         if (row->prereqs) {
1128             const struct ovsdb_idl_table_class *class = row->table->class;
1129             size_t n_columns = class->n_columns;
1130             struct json *op, *columns, *row_json;
1131             size_t idx;
1132
1133             op = json_object_create();
1134             json_array_add(operations, op);
1135             json_object_put_string(op, "op", "wait");
1136             json_object_put_string(op, "table", class->name);
1137             json_object_put(op, "timeout", json_integer_create(0));
1138             json_object_put(op, "where", where_uuid_equals(&row->uuid));
1139             json_object_put_string(op, "until", "==");
1140             columns = json_array_create_empty();
1141             json_object_put(op, "columns", columns);
1142             row_json = json_object_create();
1143             json_object_put(op, "rows", json_array_create_1(row_json));
1144
1145             BITMAP_FOR_EACH_1 (idx, n_columns, row->prereqs) {
1146                 const struct ovsdb_idl_column *column = &class->columns[idx];
1147                 json_array_add(columns, json_string_create(column->name));
1148                 json_object_put(row_json, column->name,
1149                                 ovsdb_datum_to_json(&row->old[idx],
1150                                                     &column->type));
1151             }
1152         }
1153     }
1154
1155     /* Add updates. */
1156     any_updates = false;
1157     HMAP_FOR_EACH (row, struct ovsdb_idl_row, txn_node, &txn->txn_rows) {
1158         const struct ovsdb_idl_table_class *class = row->table->class;
1159
1160         if (row->old == row->new) {
1161             continue;
1162         } else if (!row->new) {
1163             struct json *op = json_object_create();
1164             json_object_put_string(op, "op", "delete");
1165             json_object_put_string(op, "table", class->name);
1166             json_object_put(op, "where", where_uuid_equals(&row->uuid));
1167             json_array_add(operations, op);
1168             any_updates = true;
1169         } else {
1170             struct json *row_json;
1171             struct json *op;
1172             size_t idx;
1173
1174             op = json_object_create();
1175             json_object_put_string(op, "op", row->old ? "update" : "insert");
1176             json_object_put_string(op, "table", class->name);
1177             if (row->old) {
1178                 json_object_put(op, "where", where_uuid_equals(&row->uuid));
1179             } else {
1180                 struct ovsdb_idl_txn_insert *insert;
1181
1182                 json_object_put(op, "uuid-name",
1183                                 json_string_create_nocopy(
1184                                     uuid_name_from_uuid(&row->uuid)));
1185
1186                 insert = xmalloc(sizeof *insert);
1187                 insert->dummy = row->uuid;
1188                 insert->op_index = operations->u.array.n - 1;
1189                 uuid_zero(&insert->real);
1190                 hmap_insert(&txn->inserted_rows, &insert->hmap_node,
1191                             uuid_hash(&insert->dummy));
1192             }
1193             row_json = json_object_create();
1194             json_object_put(op, "row", row_json);
1195
1196             BITMAP_FOR_EACH_1 (idx, class->n_columns, row->written) {
1197                 const struct ovsdb_idl_column *column = &class->columns[idx];
1198
1199                 if (row->old
1200                     ? !ovsdb_datum_equals(&row->old[idx], &row->new[idx],
1201                                           &column->type)
1202                     : !ovsdb_datum_is_default(&row->new[idx], &column->type)) {
1203                     json_object_put(row_json, column->name,
1204                                     substitute_uuids(
1205                                         ovsdb_datum_to_json(&row->new[idx],
1206                                                             &column->type),
1207                                         txn));
1208                 }
1209             }
1210
1211             if (!row->old || !shash_is_empty(json_object(row_json))) {
1212                 json_array_add(operations, op);
1213                 any_updates = true;
1214             } else {
1215                 json_destroy(op);
1216             }
1217         }
1218     }
1219
1220     /* Add increment. */
1221     if (txn->inc_table && any_updates) {
1222         struct json *op;
1223
1224         txn->inc_index = operations->u.array.n - 1;
1225
1226         op = json_object_create();
1227         json_object_put_string(op, "op", "mutate");
1228         json_object_put_string(op, "table", txn->inc_table);
1229         json_object_put(op, "where",
1230                         substitute_uuids(json_clone(txn->inc_where), txn));
1231         json_object_put(op, "mutations",
1232                         json_array_create_1(
1233                             json_array_create_3(
1234                                 json_string_create(txn->inc_column),
1235                                 json_string_create("+="),
1236                                 json_integer_create(1))));
1237         json_array_add(operations, op);
1238
1239         op = json_object_create();
1240         json_object_put_string(op, "op", "select");
1241         json_object_put_string(op, "table", txn->inc_table);
1242         json_object_put(op, "where",
1243                         substitute_uuids(json_clone(txn->inc_where), txn));
1244         json_object_put(op, "columns",
1245                         json_array_create_1(json_string_create(
1246                                                 txn->inc_column)));
1247         json_array_add(operations, op);
1248     }
1249
1250     if (txn->comment.length) {
1251         struct json *op = json_object_create();
1252         json_object_put_string(op, "op", "comment");
1253         json_object_put_string(op, "comment", ds_cstr(&txn->comment));
1254         json_array_add(operations, op);
1255     }
1256
1257     if (txn->dry_run) {
1258         struct json *op = json_object_create();
1259         json_object_put_string(op, "op", "abort");
1260         json_array_add(operations, op);
1261     }
1262
1263     if (!any_updates) {
1264         txn->status = TXN_UNCHANGED;
1265         json_destroy(operations);
1266     } else if (!jsonrpc_session_send(
1267                    txn->idl->session,
1268                    jsonrpc_create_request(
1269                        "transact", operations, &txn->request_id))) {
1270         hmap_insert(&txn->idl->outstanding_txns, &txn->hmap_node,
1271                     json_hash(txn->request_id, 0));
1272     } else {
1273         txn->status = TXN_TRY_AGAIN;
1274     }
1275
1276     ovsdb_idl_txn_disassemble(txn);
1277     return txn->status;
1278 }
1279
1280 /* Attempts to commit 'txn', blocking until the commit either succeeds or
1281  * fails.  Returns the final commit status, which may be any TXN_* value other
1282  * than TXN_INCOMPLETE. */
1283 enum ovsdb_idl_txn_status
1284 ovsdb_idl_txn_commit_block(struct ovsdb_idl_txn *txn)
1285 {
1286     enum ovsdb_idl_txn_status status;
1287
1288     fatal_signal_run();
1289     while ((status = ovsdb_idl_txn_commit(txn)) == TXN_INCOMPLETE) {
1290         ovsdb_idl_run(txn->idl);
1291         ovsdb_idl_wait(txn->idl);
1292         ovsdb_idl_txn_wait(txn);
1293         poll_block();
1294     }
1295     return status;
1296 }
1297
1298 int64_t
1299 ovsdb_idl_txn_get_increment_new_value(const struct ovsdb_idl_txn *txn)
1300 {
1301     assert(txn->status == TXN_SUCCESS);
1302     return txn->inc_new_value;
1303 }
1304
1305 void
1306 ovsdb_idl_txn_abort(struct ovsdb_idl_txn *txn)
1307 {
1308     ovsdb_idl_txn_disassemble(txn);
1309     if (txn->status == TXN_INCOMPLETE) {
1310         txn->status = TXN_ABORTED;
1311     }
1312 }
1313
1314 const char *
1315 ovsdb_idl_txn_get_error(const struct ovsdb_idl_txn *txn)
1316 {
1317     if (txn->status != TXN_ERROR) {
1318         return ovsdb_idl_txn_status_to_string(txn->status);
1319     } else if (txn->error) {
1320         return txn->error;
1321     } else {
1322         return "no error details available";
1323     }
1324 }
1325
1326 static void
1327 ovsdb_idl_txn_set_error_json(struct ovsdb_idl_txn *txn,
1328                              const struct json *json)
1329 {
1330     if (txn->error == NULL) {
1331         txn->error = json_to_string(json, JSSF_SORT);
1332     }
1333 }
1334
1335 /* For transaction 'txn' that completed successfully, finds and returns the
1336  * permanent UUID that the database assigned to a newly inserted row, given the
1337  * 'uuid' that ovsdb_idl_txn_insert() assigned locally to that row.
1338  *
1339  * Returns NULL if 'uuid' is not a UUID assigned by ovsdb_idl_txn_insert() or
1340  * if it was assigned by that function and then deleted by
1341  * ovsdb_idl_txn_delete() within the same transaction.  (Rows that are inserted
1342  * and then deleted within a single transaction are never sent to the database
1343  * server, so it never assigns them a permanent UUID.) */
1344 const struct uuid *
1345 ovsdb_idl_txn_get_insert_uuid(const struct ovsdb_idl_txn *txn,
1346                               const struct uuid *uuid)
1347 {
1348     const struct ovsdb_idl_txn_insert *insert;
1349
1350     assert(txn->status == TXN_SUCCESS || txn->status == TXN_UNCHANGED);
1351     HMAP_FOR_EACH_IN_BUCKET (insert, struct ovsdb_idl_txn_insert, hmap_node,
1352                              uuid_hash(uuid), &txn->inserted_rows) {
1353         if (uuid_equals(uuid, &insert->dummy)) {
1354             return &insert->real;
1355         }
1356     }
1357     return NULL;
1358 }
1359
1360 static void
1361 ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
1362                        enum ovsdb_idl_txn_status status)
1363 {
1364     txn->status = status;
1365     hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
1366 }
1367
1368 void
1369 ovsdb_idl_txn_read(const struct ovsdb_idl_row *row,
1370                    const struct ovsdb_idl_column *column,
1371                    struct ovsdb_datum *datum)
1372 {
1373     const struct ovsdb_idl_table_class *class = row->table->class;
1374     size_t column_idx = column - class->columns;
1375
1376     assert(row->new != NULL);
1377     if (row->written && bitmap_is_set(row->written, column_idx)) {
1378         ovsdb_datum_clone(datum, &row->new[column_idx], &column->type);
1379     } else if (row->old) {
1380         ovsdb_datum_clone(datum, &row->old[column_idx], &column->type);
1381     } else {
1382         ovsdb_datum_init_default(datum, &column->type);
1383     }
1384 }
1385
1386 void
1387 ovsdb_idl_txn_write(const struct ovsdb_idl_row *row_,
1388                     const struct ovsdb_idl_column *column,
1389                     struct ovsdb_datum *datum)
1390 {
1391     struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
1392     const struct ovsdb_idl_table_class *class = row->table->class;
1393     size_t column_idx = column - class->columns;
1394
1395     assert(row->new != NULL);
1396     assert(column_idx < class->n_columns);
1397     if (hmap_node_is_null(&row->txn_node)) {
1398         hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
1399                     uuid_hash(&row->uuid));
1400     }
1401     if (row->old == row->new) {
1402         row->new = xmalloc(class->n_columns * sizeof *row->new);
1403     }
1404     if (!row->written) {
1405         row->written = bitmap_allocate(class->n_columns);
1406     }
1407     if (bitmap_is_set(row->written, column_idx)) {
1408         ovsdb_datum_destroy(&row->new[column_idx], &column->type);
1409     } else {
1410         bitmap_set1(row->written, column_idx);
1411     }
1412     row->new[column_idx] = *datum;
1413     (column->unparse)(row);
1414     (column->parse)(row, &row->new[column_idx]);
1415 }
1416
1417 void
1418 ovsdb_idl_txn_verify(const struct ovsdb_idl_row *row_,
1419                      const struct ovsdb_idl_column *column)
1420 {
1421     struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
1422     const struct ovsdb_idl_table_class *class = row->table->class;
1423     size_t column_idx = column - class->columns;
1424
1425     assert(row->new != NULL);
1426     if (!row->old
1427         || (row->written && bitmap_is_set(row->written, column_idx))) {
1428         return;
1429     }
1430
1431     if (hmap_node_is_null(&row->txn_node)) {
1432         hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
1433                     uuid_hash(&row->uuid));
1434     }
1435     if (!row->prereqs) {
1436         row->prereqs = bitmap_allocate(class->n_columns);
1437     }
1438     bitmap_set1(row->prereqs, column_idx);
1439 }
1440
1441 void
1442 ovsdb_idl_txn_delete(const struct ovsdb_idl_row *row_)
1443 {
1444     struct ovsdb_idl_row *row = (struct ovsdb_idl_row *) row_;
1445
1446     assert(row->new != NULL);
1447     if (!row->old) {
1448         ovsdb_idl_row_unparse(row);
1449         ovsdb_idl_row_clear_new(row);
1450         assert(!row->prereqs);
1451         hmap_remove(&row->table->rows, &row->hmap_node);
1452         hmap_remove(&row->table->idl->txn->txn_rows, &row->txn_node);
1453         free(row);
1454         return;
1455     }
1456     if (hmap_node_is_null(&row->txn_node)) {
1457         hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
1458                     uuid_hash(&row->uuid));
1459     }
1460     ovsdb_idl_row_clear_new(row);
1461     row->new = NULL;
1462 }
1463
1464 const struct ovsdb_idl_row *
1465 ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
1466                      const struct ovsdb_idl_table_class *class,
1467                      const struct uuid *uuid)
1468 {
1469     struct ovsdb_idl_row *row = ovsdb_idl_row_create__(class);
1470
1471     if (uuid) {
1472         assert(!ovsdb_idl_txn_get_row(txn, uuid));
1473         row->uuid = *uuid;
1474     } else {
1475         uuid_generate(&row->uuid);
1476     }
1477
1478     row->table = ovsdb_idl_table_from_class(txn->idl, class);
1479     row->new = xmalloc(class->n_columns * sizeof *row->new);
1480     row->written = bitmap_allocate(class->n_columns);
1481     hmap_insert(&row->table->rows, &row->hmap_node, uuid_hash(&row->uuid));
1482     hmap_insert(&txn->txn_rows, &row->txn_node, uuid_hash(&row->uuid));
1483     return row;
1484 }
1485
1486 static void
1487 ovsdb_idl_txn_abort_all(struct ovsdb_idl *idl)
1488 {
1489     struct ovsdb_idl_txn *txn;
1490
1491     HMAP_FOR_EACH (txn, struct ovsdb_idl_txn, hmap_node,
1492                    &idl->outstanding_txns) {
1493         ovsdb_idl_txn_complete(txn, TXN_TRY_AGAIN);
1494     }
1495 }
1496
1497 static struct ovsdb_idl_txn *
1498 ovsdb_idl_txn_find(struct ovsdb_idl *idl, const struct json *id)
1499 {
1500     struct ovsdb_idl_txn *txn;
1501
1502     HMAP_FOR_EACH_WITH_HASH (txn, struct ovsdb_idl_txn, hmap_node,
1503                              json_hash(id, 0), &idl->outstanding_txns) {
1504         if (json_equal(id, txn->request_id)) {
1505             return txn;
1506         }
1507     }
1508     return NULL;
1509 }
1510
1511 static bool
1512 check_json_type(const struct json *json, enum json_type type, const char *name)
1513 {
1514     if (!json) {
1515         VLOG_WARN_RL(&syntax_rl, "%s is missing", name);
1516         return false;
1517     } else if (json->type != type) {
1518         VLOG_WARN_RL(&syntax_rl, "%s is %s instead of %s",
1519                      name, json_type_to_string(json->type),
1520                      json_type_to_string(type));
1521         return false;
1522     } else {
1523         return true;
1524     }
1525 }
1526
1527 static bool
1528 ovsdb_idl_txn_process_inc_reply(struct ovsdb_idl_txn *txn,
1529                                 const struct json_array *results)
1530 {
1531     struct json *count, *rows, *row, *column;
1532     struct shash *mutate, *select;
1533
1534     if (txn->inc_index + 2 > results->n) {
1535         VLOG_WARN_RL(&syntax_rl, "reply does not contain enough operations "
1536                      "for increment (has %zu, needs %u)",
1537                      results->n, txn->inc_index + 2);
1538         return false;
1539     }
1540
1541     /* We know that this is a JSON object because the loop in
1542      * ovsdb_idl_txn_process_reply() checked. */
1543     mutate = json_object(results->elems[txn->inc_index]);
1544     count = shash_find_data(mutate, "count");
1545     if (!check_json_type(count, JSON_INTEGER, "\"mutate\" reply \"count\"")) {
1546         return false;
1547     }
1548     if (count->u.integer != 1) {
1549         VLOG_WARN_RL(&syntax_rl,
1550                      "\"mutate\" reply \"count\" is %lld instead of 1",
1551                      count->u.integer);
1552         return false;
1553     }
1554
1555     select = json_object(results->elems[txn->inc_index + 1]);
1556     rows = shash_find_data(select, "rows");
1557     if (!check_json_type(rows, JSON_ARRAY, "\"select\" reply \"rows\"")) {
1558         return false;
1559     }
1560     if (rows->u.array.n != 1) {
1561         VLOG_WARN_RL(&syntax_rl, "\"select\" reply \"rows\" has %zu elements "
1562                      "instead of 1",
1563                      rows->u.array.n);
1564         return false;
1565     }
1566     row = rows->u.array.elems[0];
1567     if (!check_json_type(row, JSON_OBJECT, "\"select\" reply row")) {
1568         return false;
1569     }
1570     column = shash_find_data(json_object(row), txn->inc_column);
1571     if (!check_json_type(column, JSON_INTEGER,
1572                          "\"select\" reply inc column")) {
1573         return false;
1574     }
1575     txn->inc_new_value = column->u.integer;
1576     return true;
1577 }
1578
1579 static bool
1580 ovsdb_idl_txn_process_insert_reply(struct ovsdb_idl_txn_insert *insert,
1581                                    const struct json_array *results)
1582 {
1583     static const struct ovsdb_base_type uuid_type = OVSDB_BASE_UUID_INIT;
1584     struct ovsdb_error *error;
1585     struct json *json_uuid;
1586     union ovsdb_atom uuid;
1587     struct shash *reply;
1588
1589     if (insert->op_index >= results->n) {
1590         VLOG_WARN_RL(&syntax_rl, "reply does not contain enough operations "
1591                      "for insert (has %zu, needs %u)",
1592                      results->n, insert->op_index);
1593         return false;
1594     }
1595
1596     /* We know that this is a JSON object because the loop in
1597      * ovsdb_idl_txn_process_reply() checked. */
1598     reply = json_object(results->elems[insert->op_index]);
1599     json_uuid = shash_find_data(reply, "uuid");
1600     if (!check_json_type(json_uuid, JSON_ARRAY, "\"insert\" reply \"uuid\"")) {
1601         return false;
1602     }
1603
1604     error = ovsdb_atom_from_json(&uuid, &uuid_type, json_uuid, NULL);
1605     if (error) {
1606         char *s = ovsdb_error_to_string(error);
1607         VLOG_WARN_RL(&syntax_rl, "\"insert\" reply \"uuid\" is not a JSON "
1608                      "UUID: %s", s);
1609         free(s);
1610         return false;
1611     }
1612
1613     insert->real = uuid.uuid;
1614
1615     return true;
1616 }
1617
1618 static bool
1619 ovsdb_idl_txn_process_reply(struct ovsdb_idl *idl,
1620                             const struct jsonrpc_msg *msg)
1621 {
1622     struct ovsdb_idl_txn *txn;
1623     enum ovsdb_idl_txn_status status;
1624
1625     txn = ovsdb_idl_txn_find(idl, msg->id);
1626     if (!txn) {
1627         return false;
1628     }
1629
1630     if (msg->type == JSONRPC_ERROR) {
1631         status = TXN_ERROR;
1632     } else if (msg->result->type != JSON_ARRAY) {
1633         VLOG_WARN_RL(&syntax_rl, "reply to \"transact\" is not JSON array");
1634         status = TXN_ERROR;
1635     } else {
1636         struct json_array *ops = &msg->result->u.array;
1637         int hard_errors = 0;
1638         int soft_errors = 0;
1639         size_t i;
1640
1641         for (i = 0; i < ops->n; i++) {
1642             struct json *op = ops->elems[i];
1643
1644             if (op->type == JSON_NULL) {
1645                 /* This isn't an error in itself but indicates that some prior
1646                  * operation failed, so make sure that we know about it. */
1647                 soft_errors++;
1648             } else if (op->type == JSON_OBJECT) {
1649                 struct json *error;
1650
1651                 error = shash_find_data(json_object(op), "error");
1652                 if (error) {
1653                     if (error->type == JSON_STRING) {
1654                         if (!strcmp(error->u.string, "timed out")) {
1655                             soft_errors++;
1656                         } else if (strcmp(error->u.string, "aborted")) {
1657                             hard_errors++;
1658                             ovsdb_idl_txn_set_error_json(txn, op);
1659                         }
1660                     } else {
1661                         hard_errors++;
1662                         ovsdb_idl_txn_set_error_json(txn, op);
1663                         VLOG_WARN_RL(&syntax_rl,
1664                                      "\"error\" in reply is not JSON string");
1665                     }
1666                 }
1667             } else {
1668                 hard_errors++;
1669                 ovsdb_idl_txn_set_error_json(txn, op);
1670                 VLOG_WARN_RL(&syntax_rl,
1671                              "operation reply is not JSON null or object");
1672             }
1673         }
1674
1675         if (!soft_errors && !hard_errors) {
1676             struct ovsdb_idl_txn_insert *insert;
1677
1678             if (txn->inc_table && !ovsdb_idl_txn_process_inc_reply(txn, ops)) {
1679                 hard_errors++;
1680             }
1681
1682             HMAP_FOR_EACH (insert, struct ovsdb_idl_txn_insert, hmap_node,
1683                            &txn->inserted_rows) {
1684                 if (!ovsdb_idl_txn_process_insert_reply(insert, ops)) {
1685                     hard_errors++;
1686                 }
1687             }
1688         }
1689
1690         status = (hard_errors ? TXN_ERROR
1691                   : soft_errors ? TXN_TRY_AGAIN
1692                   : TXN_SUCCESS);
1693     }
1694
1695     ovsdb_idl_txn_complete(txn, status);
1696     return true;
1697 }
1698
1699 struct ovsdb_idl_txn *
1700 ovsdb_idl_txn_get(const struct ovsdb_idl_row *row)
1701 {
1702     struct ovsdb_idl_txn *txn = row->table->idl->txn;
1703     assert(txn != NULL);
1704     return txn;
1705 }
1706
1707 struct ovsdb_idl *
1708 ovsdb_idl_txn_get_idl (struct ovsdb_idl_txn *txn)
1709 {
1710     return txn->idl;
1711 }
1712