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