ovsdb/execution.c

   1 /* Copyright (c) 2009, 2010, 2011, 2012, 2013 Nicira, Inc.
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <config.h>
  17
  18 #include <limits.h>
  19
  20 #include "column.h"
  21 #include "condition.h"
  22 #include "file.h"
  23 #include "json.h"
  24 #include "mutation.h"
  25 #include "ovsdb-data.h"
  26 #include "ovsdb-error.h"
  27 #include "ovsdb-parser.h"
  28 #include "ovsdb.h"
  29 #include "query.h"
  30 #include "row.h"
  31 #include "server.h"
  32 #include "table.h"
  33 #include "timeval.h"
  34 #include "transaction.h"
  35
  36 struct ovsdb_execution {
  37     struct ovsdb *db;
  38     const struct ovsdb_session *session;
  39     struct ovsdb_txn *txn;
  40     struct ovsdb_symbol_table *symtab;
  41     bool durable;
  42
  43     /* Triggers. */
  44     long long int elapsed_msec;
  45     long long int timeout_msec;
  46 };
  47
  48 typedef struct ovsdb_error *ovsdb_operation_executor(struct ovsdb_execution *,
  49                                                      struct ovsdb_parser *,
  50                                                      struct json *result);
  51
  52 static ovsdb_operation_executor ovsdb_execute_insert;
  53 static ovsdb_operation_executor ovsdb_execute_select;
  54 static ovsdb_operation_executor ovsdb_execute_update;
  55 static ovsdb_operation_executor ovsdb_execute_mutate;
  56 static ovsdb_operation_executor ovsdb_execute_delete;
  57 static ovsdb_operation_executor ovsdb_execute_wait;
  58 static ovsdb_operation_executor ovsdb_execute_commit;
  59 static ovsdb_operation_executor ovsdb_execute_abort;
  60 static ovsdb_operation_executor ovsdb_execute_comment;
  61 static ovsdb_operation_executor ovsdb_execute_assert;
  62
  63 static ovsdb_operation_executor *
  64 lookup_executor(const char *name)
  65 {
  66     struct ovsdb_operation {
  67         const char *name;
  68         ovsdb_operation_executor *executor;
  69     };
  70
  71     static const struct ovsdb_operation operations[] = {
  72         { "insert", ovsdb_execute_insert },
  73         { "select", ovsdb_execute_select },
  74         { "update", ovsdb_execute_update },
  75         { "mutate", ovsdb_execute_mutate },
  76         { "delete", ovsdb_execute_delete },
  77         { "wait", ovsdb_execute_wait },
  78         { "commit", ovsdb_execute_commit },
  79         { "abort", ovsdb_execute_abort },
  80         { "comment", ovsdb_execute_comment },
  81         { "assert", ovsdb_execute_assert },
  82     };
  83
  84     size_t i;
  85
  86     for (i = 0; i < ARRAY_SIZE(operations); i++) {
  87         const struct ovsdb_operation *c = &operations[i];
  88         if (!strcmp(c->name, name)) {
  89             return c->executor;
  90         }
  91     }
  92     return NULL;
  93 }
  94
  95 struct json *
  96 ovsdb_execute(struct ovsdb *db, const struct ovsdb_session *session,
  97               const struct json *params,
  98               long long int elapsed_msec, long long int *timeout_msec)
  99 {
 100     struct ovsdb_execution x;
 101     struct ovsdb_error *error;
 102     struct json *results;
 103     size_t n_operations;
 104     size_t i;
 105
 106     if (params->type != JSON_ARRAY
 107         || !params->u.array.n
 108         || params->u.array.elems[0]->type != JSON_STRING
 109         || strcmp(params->u.array.elems[0]->u.string, db->schema->name)) {
 110         if (params->type != JSON_ARRAY) {
 111             error = ovsdb_syntax_error(params, NULL, "array expected");
 112         } else {
 113             error = ovsdb_syntax_error(params, NULL, "database name expected "
 114                                        "as first parameter");
 115         }
 116
 117         results = ovsdb_error_to_json(error);
 118         ovsdb_error_destroy(error);
 119         return results;
 120     }
 121
 122     x.db = db;
 123     x.session = session;
 124     x.txn = ovsdb_txn_create(db);
 125     x.symtab = ovsdb_symbol_table_create();
 126     x.durable = false;
 127     x.elapsed_msec = elapsed_msec;
 128     x.timeout_msec = LLONG_MAX;
 129     results = NULL;
 130
 131     results = json_array_create_empty();
 132     n_operations = params->u.array.n - 1;
 133     error = NULL;
 134     for (i = 1; i <= n_operations; i++) {
 135         struct json *operation = params->u.array.elems[i];
 136         struct ovsdb_error *parse_error;
 137         struct ovsdb_parser parser;
 138         struct json *result;
 139         const struct json *op;
 140
 141         /* Parse and execute operation. */
 142         ovsdb_parser_init(&parser, operation,
 143                           "ovsdb operation %"PRIuSIZE" of %"PRIuSIZE, i, n_operations);
 144         op = ovsdb_parser_member(&parser, "op", OP_ID);
 145         result = json_object_create();
 146         if (op) {
 147             const char *op_name = json_string(op);
 148             ovsdb_operation_executor *executor = lookup_executor(op_name);
 149             if (executor) {
 150                 error = executor(&x, &parser, result);
 151             } else {
 152                 ovsdb_parser_raise_error(&parser, "No operation \"%s\"",
 153                                          op_name);
 154             }
 155         } else {
 156             ovs_assert(ovsdb_parser_has_error(&parser));
 157         }
 158
 159         /* A parse error overrides any other error.
 160          * An error overrides any other result. */
 161         parse_error = ovsdb_parser_finish(&parser);
 162         if (parse_error) {
 163             ovsdb_error_destroy(error);
 164             error = parse_error;
 165         }
 166         if (error) {
 167             json_destroy(result);
 168             result = ovsdb_error_to_json(error);
 169         }
 170         if (error && !strcmp(ovsdb_error_get_tag(error), "not supported")
 171             && timeout_msec) {
 172             ovsdb_txn_abort(x.txn);
 173             *timeout_msec = x.timeout_msec;
 174
 175             json_destroy(result);
 176             json_destroy(results);
 177             results = NULL;
 178             goto exit;
 179         }
 180
 181         /* Add result to array. */
 182         json_array_add(results, result);
 183         if (error) {
 184             break;
 185         }
 186     }
 187
 188     if (!error) {
 189         error = ovsdb_txn_commit(x.txn, x.durable);
 190         if (error) {
 191             json_array_add(results, ovsdb_error_to_json(error));
 192         }
 193     } else {
 194         ovsdb_txn_abort(x.txn);
 195     }
 196
 197     while (json_array(results)->n < n_operations) {
 198         json_array_add(results, json_null_create());
 199     }
 200
 201 exit:
 202     ovsdb_error_destroy(error);
 203     ovsdb_symbol_table_destroy(x.symtab);
 204
 205     return results;
 206 }
 207
 208 static struct ovsdb_error *
 209 ovsdb_execute_commit(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 210                      struct json *result OVS_UNUSED)
 211 {
 212     const struct json *durable;
 213
 214     durable = ovsdb_parser_member(parser, "durable", OP_BOOLEAN);
 215     if (durable && json_boolean(durable)) {
 216         x->durable = true;
 217     }
 218     return NULL;
 219 }
 220
 221 static struct ovsdb_error *
 222 ovsdb_execute_abort(struct ovsdb_execution *x OVS_UNUSED,
 223                     struct ovsdb_parser *parser OVS_UNUSED,
 224                     struct json *result OVS_UNUSED)
 225 {
 226     return ovsdb_error("aborted", "aborted by request");
 227 }
 228
 229 static struct ovsdb_table *
 230 parse_table(struct ovsdb_execution *x,
 231             struct ovsdb_parser *parser, const char *member)
 232 {
 233     struct ovsdb_table *table;
 234     const char *table_name;
 235     const struct json *json;
 236
 237     json = ovsdb_parser_member(parser, member, OP_ID);
 238     if (!json) {
 239         return NULL;
 240     }
 241     table_name = json_string(json);
 242
 243     table = shash_find_data(&x->db->tables, table_name);
 244     if (!table) {
 245         ovsdb_parser_raise_error(parser, "No table named %s.", table_name);
 246     }
 247     return table;
 248 }
 249
 250 static OVS_WARN_UNUSED_RESULT struct ovsdb_error *
 251 parse_row(const struct json *json, const struct ovsdb_table *table,
 252           struct ovsdb_symbol_table *symtab,
 253           struct ovsdb_row **rowp, struct ovsdb_column_set *columns)
 254 {
 255     struct ovsdb_error *error;
 256     struct ovsdb_row *row;
 257
 258     *rowp = NULL;
 259
 260     if (!table) {
 261         return OVSDB_BUG("null table");
 262     }
 263     if (!json) {
 264         return OVSDB_BUG("null row");
 265     }
 266
 267     row = ovsdb_row_create(table);
 268     error = ovsdb_row_from_json(row, json, symtab, columns);
 269     if (error) {
 270         ovsdb_row_destroy(row);
 271         return error;
 272     } else {
 273         *rowp = row;
 274         return NULL;
 275     }
 276 }
 277
 278 static struct ovsdb_error *
 279 ovsdb_execute_insert(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 280                      struct json *result)
 281 {
 282     struct ovsdb_table *table;
 283     struct ovsdb_row *row = NULL;
 284     const struct json *uuid_name, *row_json;
 285     struct ovsdb_error *error;
 286     struct uuid row_uuid;
 287
 288     table = parse_table(x, parser, "table");
 289     uuid_name = ovsdb_parser_member(parser, "uuid-name", OP_ID | OP_OPTIONAL);
 290     row_json = ovsdb_parser_member(parser, "row", OP_OBJECT);
 291     error = ovsdb_parser_get_error(parser);
 292     if (error) {
 293         return error;
 294     }
 295
 296     if (uuid_name) {
 297         struct ovsdb_symbol *symbol;
 298
 299         symbol = ovsdb_symbol_table_insert(x->symtab, json_string(uuid_name));
 300         if (symbol->created) {
 301             return ovsdb_syntax_error(uuid_name, "duplicate uuid-name",
 302                                       "This \"uuid-name\" appeared on an "
 303                                       "earlier \"insert\" operation.");
 304         }
 305         row_uuid = symbol->uuid;
 306         symbol->created = true;
 307     } else {
 308         uuid_generate(&row_uuid);
 309     }
 310
 311     if (!error) {
 312         error = parse_row(row_json, table, x->symtab, &row, NULL);
 313     }
 314     if (!error) {
 315         /* Check constraints for columns not included in "row", in case the
 316          * default values do not satisfy the constraints.  We could check only
 317          * the columns that have their default values by supplying an
 318          * ovsdb_column_set to parse_row() above, but I suspect that this is
 319          * cheaper.  */
 320         const struct shash_node *node;
 321
 322         SHASH_FOR_EACH (node, &table->schema->columns) {
 323             const struct ovsdb_column *column = node->data;
 324             const struct ovsdb_datum *datum = &row->fields[column->index];
 325
 326             /* If there are 0 keys or pairs, there's nothing to check.
 327              * If there is 1, it might be a default value.
 328              * If there are more, it can't be a default value, so the value has
 329              * already been checked. */
 330             if (datum->n == 1) {
 331                 error = ovsdb_datum_check_constraints(datum, &column->type);
 332                 if (error) {
 333                     ovsdb_row_destroy(row);
 334                     break;
 335                 }
 336             }
 337         }
 338     }
 339     if (!error) {
 340         *ovsdb_row_get_uuid_rw(row) = row_uuid;
 341         ovsdb_txn_row_insert(x->txn, row);
 342         json_object_put(result, "uuid",
 343                         ovsdb_datum_to_json(&row->fields[OVSDB_COL_UUID],
 344                                             &ovsdb_type_uuid));
 345     }
 346     return error;
 347 }
 348
 349 static struct ovsdb_error *
 350 ovsdb_execute_select(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 351                      struct json *result)
 352 {
 353     struct ovsdb_table *table;
 354     const struct json *where, *columns_json, *sort_json;
 355     struct ovsdb_condition condition = OVSDB_CONDITION_INITIALIZER(&condition);
 356     struct ovsdb_column_set columns = OVSDB_COLUMN_SET_INITIALIZER;
 357     struct ovsdb_column_set sort = OVSDB_COLUMN_SET_INITIALIZER;
 358     struct ovsdb_error *error;
 359
 360     table = parse_table(x, parser, "table");
 361     where = ovsdb_parser_member(parser, "where", OP_ARRAY);
 362     columns_json = ovsdb_parser_member(parser, "columns",
 363                                        OP_ARRAY | OP_OPTIONAL);
 364     sort_json = ovsdb_parser_member(parser, "sort", OP_ARRAY | OP_OPTIONAL);
 365
 366     error = ovsdb_parser_get_error(parser);
 367     if (!error) {
 368         error = ovsdb_condition_from_json(table->schema, where, x->symtab,
 369                                           &condition);
 370     }
 371     if (!error) {
 372         error = ovsdb_column_set_from_json(columns_json, table->schema,
 373                                            &columns);
 374     }
 375     if (!error) {
 376         error = ovsdb_column_set_from_json(sort_json, table->schema, &sort);
 377     }
 378     if (!error) {
 379         struct ovsdb_row_set rows = OVSDB_ROW_SET_INITIALIZER;
 380
 381         ovsdb_query_distinct(table, &condition, &columns, &rows);
 382         ovsdb_row_set_sort(&rows, &sort);
 383         json_object_put(result, "rows",
 384                         ovsdb_row_set_to_json(&rows, &columns));
 385
 386         ovsdb_row_set_destroy(&rows);
 387     }
 388
 389     ovsdb_column_set_destroy(&columns);
 390     ovsdb_column_set_destroy(&sort);
 391     ovsdb_condition_destroy(&condition);
 392
 393     return error;
 394 }
 395
 396 struct update_row_cbdata {
 397     size_t n_matches;
 398     struct ovsdb_txn *txn;
 399     const struct ovsdb_row *row;
 400     const struct ovsdb_column_set *columns;
 401 };
 402
 403 static bool
 404 update_row_cb(const struct ovsdb_row *row, void *ur_)
 405 {
 406     struct update_row_cbdata *ur = ur_;
 407
 408     ur->n_matches++;
 409     if (!ovsdb_row_equal_columns(row, ur->row, ur->columns)) {
 410         ovsdb_row_update_columns(ovsdb_txn_row_modify(ur->txn, row),
 411                                  ur->row, ur->columns);
 412     }
 413
 414     return true;
 415 }
 416
 417 static struct ovsdb_error *
 418 ovsdb_execute_update(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 419                      struct json *result)
 420 {
 421     struct ovsdb_table *table;
 422     const struct json *where, *row_json;
 423     struct ovsdb_condition condition = OVSDB_CONDITION_INITIALIZER(&condition);
 424     struct ovsdb_column_set columns = OVSDB_COLUMN_SET_INITIALIZER;
 425     struct ovsdb_row *row = NULL;
 426     struct update_row_cbdata ur;
 427     struct ovsdb_error *error;
 428
 429     table = parse_table(x, parser, "table");
 430     where = ovsdb_parser_member(parser, "where", OP_ARRAY);
 431     row_json = ovsdb_parser_member(parser, "row", OP_OBJECT);
 432     error = ovsdb_parser_get_error(parser);
 433     if (!error) {
 434         error = parse_row(row_json, table, x->symtab, &row, &columns);
 435     }
 436     if (!error) {
 437         size_t i;
 438
 439         for (i = 0; i < columns.n_columns; i++) {
 440             const struct ovsdb_column *column = columns.columns[i];
 441
 442             if (!column->mutable) {
 443                 error = ovsdb_syntax_error(parser->json,
 444                                            "constraint violation",
 445                                            "Cannot update immutable column %s "
 446                                            "in table %s.",
 447                                            column->name, table->schema->name);
 448                 break;
 449             }
 450         }
 451     }
 452     if (!error) {
 453         error = ovsdb_condition_from_json(table->schema, where, x->symtab,
 454                                           &condition);
 455     }
 456     if (!error) {
 457         ur.n_matches = 0;
 458         ur.txn = x->txn;
 459         ur.row = row;
 460         ur.columns = &columns;
 461         ovsdb_query(table, &condition, update_row_cb, &ur);
 462         json_object_put(result, "count", json_integer_create(ur.n_matches));
 463     }
 464
 465     ovsdb_row_destroy(row);
 466     ovsdb_column_set_destroy(&columns);
 467     ovsdb_condition_destroy(&condition);
 468
 469     return error;
 470 }
 471
 472 struct mutate_row_cbdata {
 473     size_t n_matches;
 474     struct ovsdb_txn *txn;
 475     const struct ovsdb_mutation_set *mutations;
 476     struct ovsdb_error **error;
 477 };
 478
 479 static bool
 480 mutate_row_cb(const struct ovsdb_row *row, void *mr_)
 481 {
 482     struct mutate_row_cbdata *mr = mr_;
 483
 484     mr->n_matches++;
 485     *mr->error = ovsdb_mutation_set_execute(ovsdb_txn_row_modify(mr->txn, row),
 486                                             mr->mutations);
 487     return *mr->error == NULL;
 488 }
 489
 490 static struct ovsdb_error *
 491 ovsdb_execute_mutate(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 492                      struct json *result)
 493 {
 494     struct ovsdb_table *table;
 495     const struct json *where;
 496     const struct json *mutations_json;
 497     struct ovsdb_condition condition = OVSDB_CONDITION_INITIALIZER(&condition);
 498     struct ovsdb_mutation_set mutations = OVSDB_MUTATION_SET_INITIALIZER;
 499     struct ovsdb_row *row = NULL;
 500     struct mutate_row_cbdata mr;
 501     struct ovsdb_error *error;
 502
 503     table = parse_table(x, parser, "table");
 504     where = ovsdb_parser_member(parser, "where", OP_ARRAY);
 505     mutations_json = ovsdb_parser_member(parser, "mutations", OP_ARRAY);
 506     error = ovsdb_parser_get_error(parser);
 507     if (!error) {
 508         error = ovsdb_mutation_set_from_json(table->schema, mutations_json,
 509                                              x->symtab, &mutations);
 510     }
 511     if (!error) {
 512         error = ovsdb_condition_from_json(table->schema, where, x->symtab,
 513                                           &condition);
 514     }
 515     if (!error) {
 516         mr.n_matches = 0;
 517         mr.txn = x->txn;
 518         mr.mutations = &mutations;
 519         mr.error = &error;
 520         ovsdb_query(table, &condition, mutate_row_cb, &mr);
 521         json_object_put(result, "count", json_integer_create(mr.n_matches));
 522     }
 523
 524     ovsdb_row_destroy(row);
 525     ovsdb_mutation_set_destroy(&mutations);
 526     ovsdb_condition_destroy(&condition);
 527
 528     return error;
 529 }
 530
 531 struct delete_row_cbdata {
 532     size_t n_matches;
 533     const struct ovsdb_table *table;
 534     struct ovsdb_txn *txn;
 535 };
 536
 537 static bool
 538 delete_row_cb(const struct ovsdb_row *row, void *dr_)
 539 {
 540     struct delete_row_cbdata *dr = dr_;
 541
 542     dr->n_matches++;
 543     ovsdb_txn_row_delete(dr->txn, row);
 544
 545     return true;
 546 }
 547
 548 static struct ovsdb_error *
 549 ovsdb_execute_delete(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 550                      struct json *result)
 551 {
 552     struct ovsdb_table *table;
 553     const struct json *where;
 554     struct ovsdb_condition condition = OVSDB_CONDITION_INITIALIZER(&condition);
 555     struct ovsdb_error *error;
 556
 557     where = ovsdb_parser_member(parser, "where", OP_ARRAY);
 558     table = parse_table(x, parser, "table");
 559     error = ovsdb_parser_get_error(parser);
 560     if (!error) {
 561         error = ovsdb_condition_from_json(table->schema, where, x->symtab,
 562                                           &condition);
 563     }
 564     if (!error) {
 565         struct delete_row_cbdata dr;
 566
 567         dr.n_matches = 0;
 568         dr.table = table;
 569         dr.txn = x->txn;
 570         ovsdb_query(table, &condition, delete_row_cb, &dr);
 571
 572         json_object_put(result, "count", json_integer_create(dr.n_matches));
 573     }
 574
 575     ovsdb_condition_destroy(&condition);
 576
 577     return error;
 578 }
 579
 580 struct wait_auxdata {
 581     struct ovsdb_row_hash *actual;
 582     struct ovsdb_row_hash *expected;
 583     bool *equal;
 584 };
 585
 586 static bool
 587 ovsdb_execute_wait_query_cb(const struct ovsdb_row *row, void *aux_)
 588 {
 589     struct wait_auxdata *aux = aux_;
 590
 591     if (ovsdb_row_hash_contains(aux->expected, row)) {
 592         ovsdb_row_hash_insert(aux->actual, row);
 593         return true;
 594     } else {
 595         /* The query row isn't in the expected result set, so the actual and
 596          * expected results sets definitely differ and we can short-circuit the
 597          * rest of the query. */
 598         *aux->equal = false;
 599         return false;
 600     }
 601 }
 602
 603 static struct ovsdb_error *
 604 ovsdb_execute_wait(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 605                    struct json *result OVS_UNUSED)
 606 {
 607     struct ovsdb_table *table;
 608     const struct json *timeout, *where, *columns_json, *until, *rows;
 609     struct ovsdb_condition condition = OVSDB_CONDITION_INITIALIZER(&condition);
 610     struct ovsdb_column_set columns = OVSDB_COLUMN_SET_INITIALIZER;
 611     struct ovsdb_row_hash expected = OVSDB_ROW_HASH_INITIALIZER(expected);
 612     struct ovsdb_row_hash actual = OVSDB_ROW_HASH_INITIALIZER(actual);
 613     struct ovsdb_error *error;
 614     struct wait_auxdata aux;
 615     long long int timeout_msec = 0;
 616     size_t i;
 617
 618     timeout = ovsdb_parser_member(parser, "timeout", OP_NUMBER | OP_OPTIONAL);
 619     where = ovsdb_parser_member(parser, "where", OP_ARRAY);
 620     columns_json = ovsdb_parser_member(parser, "columns",
 621                                        OP_ARRAY | OP_OPTIONAL);
 622     until = ovsdb_parser_member(parser, "until", OP_STRING);
 623     rows = ovsdb_parser_member(parser, "rows", OP_ARRAY);
 624     table = parse_table(x, parser, "table");
 625     error = ovsdb_parser_get_error(parser);
 626     if (!error) {
 627         error = ovsdb_condition_from_json(table->schema, where, x->symtab,
 628                                           &condition);
 629     }
 630     if (!error) {
 631         error = ovsdb_column_set_from_json(columns_json, table->schema,
 632                                            &columns);
 633     }
 634     if (!error) {
 635         if (timeout) {
 636             timeout_msec = MIN(LLONG_MAX, json_real(timeout));
 637             if (timeout_msec < 0) {
 638                 error = ovsdb_syntax_error(timeout, NULL,
 639                                            "timeout must be nonnegative");
 640             } else if (timeout_msec < x->timeout_msec) {
 641                 x->timeout_msec = timeout_msec;
 642             }
 643         } else {
 644             timeout_msec = LLONG_MAX;
 645         }
 646     }
 647     if (!error) {
 648         if (strcmp(json_string(until), "==")
 649             && strcmp(json_string(until), "!=")) {
 650             error = ovsdb_syntax_error(until, NULL,
 651                                        "\"until\" must be \"==\" or \"!=\"");
 652         }
 653     }
 654     if (!error) {
 655         /* Parse "rows" into 'expected'. */
 656         ovsdb_row_hash_init(&expected, &columns);
 657         for (i = 0; i < rows->u.array.n; i++) {
 658             struct ovsdb_row *row;
 659
 660             row = ovsdb_row_create(table);
 661             error = ovsdb_row_from_json(row, rows->u.array.elems[i], x->symtab,
 662                                         NULL);
 663             if (error) {
 664                 ovsdb_row_destroy(row);
 665                 break;
 666             }
 667
 668             if (!ovsdb_row_hash_insert(&expected, row)) {
 669                 /* XXX Perhaps we should abort with an error or log a
 670                  * warning. */
 671                 ovsdb_row_destroy(row);
 672             }
 673         }
 674     }
 675     if (!error) {
 676         /* Execute query. */
 677         bool equal = true;
 678         ovsdb_row_hash_init(&actual, &columns);
 679         aux.actual = &actual;
 680         aux.expected = &expected;
 681         aux.equal = &equal;
 682         ovsdb_query(table, &condition, ovsdb_execute_wait_query_cb, &aux);
 683         if (equal) {
 684             /* We know that every row in 'actual' is also in 'expected'.  We
 685              * also know that all of the rows in 'actual' are distinct and that
 686              * all of the rows in 'expected' are distinct.  Therefore, if
 687              * 'actual' and 'expected' have the same number of rows, then they
 688              * have the same content. */
 689             size_t n_actual = ovsdb_row_hash_count(&actual);
 690             size_t n_expected = ovsdb_row_hash_count(&expected);
 691             equal = n_actual == n_expected;
 692         }
 693         if (!strcmp(json_string(until), "==") != equal) {
 694             if (timeout && x->elapsed_msec >= timeout_msec) {
 695                 if (x->elapsed_msec) {
 696                     error = ovsdb_error("timed out",
 697                                         "\"wait\" timed out after %lld ms",
 698                                         x->elapsed_msec);
 699                 } else {
 700                     error = ovsdb_error("timed out", "\"wait\" timed out");
 701                 }
 702             } else {
 703                 /* ovsdb_execute() will change this, if triggers really are
 704                  * supported. */
 705                 error = ovsdb_error("not supported", "triggers not supported");
 706             }
 707         }
 708     }
 709
 710
 711     ovsdb_row_hash_destroy(&expected, true);
 712     ovsdb_row_hash_destroy(&actual, false);
 713     ovsdb_column_set_destroy(&columns);
 714     ovsdb_condition_destroy(&condition);
 715
 716     return error;
 717 }
 718
 719 static struct ovsdb_error *
 720 ovsdb_execute_comment(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 721                       struct json *result OVS_UNUSED)
 722 {
 723     const struct json *comment;
 724
 725     comment = ovsdb_parser_member(parser, "comment", OP_STRING);
 726     if (!comment) {
 727         return NULL;
 728     }
 729     ovsdb_txn_add_comment(x->txn, json_string(comment));
 730
 731     return NULL;
 732 }
 733
 734 static struct ovsdb_error *
 735 ovsdb_execute_assert(struct ovsdb_execution *x, struct ovsdb_parser *parser,
 736                      struct json *result OVS_UNUSED)
 737 {
 738     const struct json *lock_name;
 739
 740     lock_name = ovsdb_parser_member(parser, "lock", OP_ID);
 741     if (!lock_name) {
 742         return NULL;
 743     }
 744
 745     if (x->session) {
 746         const struct ovsdb_lock_waiter *waiter;
 747
 748         waiter = ovsdb_session_get_lock_waiter(x->session,
 749                                                json_string(lock_name));
 750         if (waiter && ovsdb_lock_waiter_is_owner(waiter)) {
 751             return NULL;
 752         }
 753     }
 754
 755     return ovsdb_error("not owner", "Asserted lock %s not held.",
 756                        json_string(lock_name));
 757 }