ovsdb/execution.c

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