ovsdb/monitor.c

   1 /*
   2  * Copyright (c) 2015 Nicira, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <config.h>
  18
  19 #include <errno.h>
  20
  21 #include "bitmap.h"
  22 #include "column.h"
  23 #include "openvswitch/dynamic-string.h"
  24 #include "openvswitch/json.h"
  25 #include "jsonrpc.h"
  26 #include "ovsdb-error.h"
  27 #include "ovsdb-parser.h"
  28 #include "ovsdb.h"
  29 #include "row.h"
  30 #include "condition.h"
  31 #include "simap.h"
  32 #include "hash.h"
  33 #include "table.h"
  34 #include "hash.h"
  35 #include "timeval.h"
  36 #include "transaction.h"
  37 #include "jsonrpc-server.h"
  38 #include "monitor.h"
  39 #include "util.h"
  40 #include "openvswitch/vlog.h"
  41
  42 VLOG_DEFINE_THIS_MODULE(ovsdb_monitor);
  43
  44 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class;
  45 static struct hmap ovsdb_monitors = HMAP_INITIALIZER(&ovsdb_monitors);
  46
  47 /* Keep state of session's conditions */
  48 struct ovsdb_monitor_session_condition {
  49     bool conditional;
  50     size_t n_true_cnd;
  51     struct shash tables;     /* Contains
  52                               *   "struct ovsdb_monitor_table_condition *"s. */
  53 };
  54
  55 /* Monitored table session's conditions */
  56 struct ovsdb_monitor_table_condition {
  57     const struct ovsdb_table *table;
  58     struct ovsdb_monitor_table *mt;
  59     struct ovsdb_condition old_condition;
  60     struct ovsdb_condition new_condition;
  61 };
  62
  63 /*  Backend monitor.
  64  *
  65  *  ovsdb_monitor keep track of the ovsdb changes.
  66  */
  67
  68 /* A collection of tables being monitored. */
  69 struct ovsdb_monitor {
  70     struct ovsdb_replica replica;
  71     struct shash tables;     /* Holds "struct ovsdb_monitor_table"s. */
  72     struct ovs_list jsonrpc_monitors;  /* Contains "jsonrpc_monitor_node"s. */
  73     struct ovsdb *db;
  74     uint64_t n_transactions;      /* Count number of committed transactions. */
  75     struct hmap_node hmap_node;   /* Elements within ovsdb_monitors.  */
  76     struct hmap json_cache;       /* Contains "ovsdb_monitor_json_cache_node"s.*/
  77 };
  78
  79 /* A json object of updates between 'from_txn' and 'dbmon->n_transactions'
  80  * inclusive.  */
  81 struct ovsdb_monitor_json_cache_node {
  82     struct hmap_node hmap_node;   /* Elements in json cache. */
  83     enum ovsdb_monitor_version version;
  84     uint64_t from_txn;
  85     struct json *json;            /* Null, or a cloned of json */
  86 };
  87
  88 struct jsonrpc_monitor_node {
  89     struct ovsdb_jsonrpc_monitor *jsonrpc_monitor;
  90     struct ovs_list node;
  91 };
  92
  93 /* A particular column being monitored. */
  94 struct ovsdb_monitor_column {
  95     const struct ovsdb_column *column;
  96     enum ovsdb_monitor_selection select;
  97     bool monitored;
  98 };
  99
 100 /* A row that has changed in a monitored table. */
 101 struct ovsdb_monitor_row {
 102     struct hmap_node hmap_node; /* In ovsdb_jsonrpc_monitor_table.changes. */
 103     struct uuid uuid;           /* UUID of row that changed. */
 104     struct ovsdb_datum *old;    /* Old data, NULL for an inserted row. */
 105     struct ovsdb_datum *new;    /* New data, NULL for a deleted row. */
 106 };
 107
 108 /* Contains 'struct ovsdb_monitor_row's for rows that have been
 109  * updated but not yet flushed to all the jsonrpc connection.
 110  *
 111  * 'n_refs' represent the number of jsonrpc connections that have
 112  * not received updates. Generate the update for the last jsonprc
 113  * connection will also destroy the whole "struct ovsdb_monitor_changes"
 114  * object.
 115  *
 116  * 'transaction' stores the first update's transaction id.
 117  * */
 118 struct ovsdb_monitor_changes {
 119     struct ovsdb_monitor_table *mt;
 120     struct hmap rows;
 121     int n_refs;
 122     uint64_t transaction;
 123     struct hmap_node hmap_node;  /* Element in ovsdb_monitor_tables' changes
 124                                     hmap.  */
 125 };
 126
 127 /* A particular table being monitored. */
 128 struct ovsdb_monitor_table {
 129     const struct ovsdb_table *table;
 130
 131     /* This is the union (bitwise-OR) of the 'select' values in all of the
 132      * members of 'columns' below. */
 133     enum ovsdb_monitor_selection select;
 134
 135     /* Columns being monitored. */
 136     struct ovsdb_monitor_column *columns;
 137     size_t n_columns;
 138     size_t n_monitored_columns;
 139     size_t allocated_columns;
 140
 141     /* Columns in ovsdb_monitor_row have different indexes then in
 142      * ovsdb_row. This field maps between column->index to the index in the
 143      * ovsdb_monitor_row. It is used for condition evaluation. */
 144     unsigned int *columns_index_map;
 145
 146     /* Contains 'ovsdb_monitor_changes' indexed by 'transaction'. */
 147     struct hmap changes;
 148 };
 149
 150 enum ovsdb_monitor_row_type {
 151     OVSDB_ROW,
 152     OVSDB_MONITOR_ROW
 153 };
 154
 155 typedef struct json *
 156 (*compose_row_update_cb_func)
 157     (const struct ovsdb_monitor_table *mt,
 158      const struct ovsdb_monitor_session_condition * condition,
 159      enum ovsdb_monitor_row_type row_type,
 160      const void *,
 161      bool initial, unsigned long int *changed);
 162
 163 static void ovsdb_monitor_destroy(struct ovsdb_monitor *dbmon);
 164 static struct ovsdb_monitor_changes * ovsdb_monitor_table_add_changes(
 165     struct ovsdb_monitor_table *mt, uint64_t next_txn);
 166 static struct ovsdb_monitor_changes *ovsdb_monitor_table_find_changes(
 167     struct ovsdb_monitor_table *mt, uint64_t unflushed);
 168 static void ovsdb_monitor_changes_destroy(
 169                                   struct ovsdb_monitor_changes *changes);
 170 static void ovsdb_monitor_table_track_changes(struct ovsdb_monitor_table *mt,
 171                                   uint64_t unflushed);
 172
 173 static uint32_t
 174 json_cache_hash(enum ovsdb_monitor_version version, uint64_t from_txn)
 175 {
 176     uint32_t hash;
 177
 178     hash = hash_uint64(version);
 179     hash = hash_uint64_basis(from_txn, hash);
 180
 181     return hash;
 182 }
 183
 184 static struct ovsdb_monitor_json_cache_node *
 185 ovsdb_monitor_json_cache_search(const struct ovsdb_monitor *dbmon,
 186                                 enum ovsdb_monitor_version version,
 187                                 uint64_t from_txn)
 188 {
 189     struct ovsdb_monitor_json_cache_node *node;
 190     uint32_t hash = json_cache_hash(version, from_txn);
 191
 192     HMAP_FOR_EACH_WITH_HASH(node, hmap_node, hash, &dbmon->json_cache) {
 193         if (node->from_txn == from_txn && node->version == version) {
 194             return node;
 195         }
 196     }
 197
 198     return NULL;
 199 }
 200
 201 static void
 202 ovsdb_monitor_json_cache_insert(struct ovsdb_monitor *dbmon,
 203                                 enum ovsdb_monitor_version version,
 204                                 uint64_t from_txn, struct json *json)
 205 {
 206     struct ovsdb_monitor_json_cache_node *node;
 207     uint32_t hash = json_cache_hash(version, from_txn);
 208
 209     node = xmalloc(sizeof *node);
 210
 211     node->version = version;
 212     node->from_txn = from_txn;
 213     node->json = json ? json_clone(json) : NULL;
 214
 215     hmap_insert(&dbmon->json_cache, &node->hmap_node, hash);
 216 }
 217
 218 static void
 219 ovsdb_monitor_json_cache_flush(struct ovsdb_monitor *dbmon)
 220 {
 221     struct ovsdb_monitor_json_cache_node *node;
 222
 223     HMAP_FOR_EACH_POP(node, hmap_node, &dbmon->json_cache) {
 224         json_destroy(node->json);
 225         free(node);
 226     }
 227 }
 228
 229 static int
 230 compare_ovsdb_monitor_column(const void *a_, const void *b_)
 231 {
 232     const struct ovsdb_monitor_column *a = a_;
 233     const struct ovsdb_monitor_column *b = b_;
 234
 235     /* put all monitored columns at the begining */
 236     if (a->monitored != b->monitored) {
 237         return a->monitored ? -1 : 1;
 238     }
 239
 240     return a->column < b->column ? -1 : a->column > b->column;
 241 }
 242
 243 static struct ovsdb_monitor *
 244 ovsdb_monitor_cast(struct ovsdb_replica *replica)
 245 {
 246     ovs_assert(replica->class == &ovsdb_jsonrpc_replica_class);
 247     return CONTAINER_OF(replica, struct ovsdb_monitor, replica);
 248 }
 249
 250 /* Finds and returns the ovsdb_monitor_row in 'mt->changes->rows' for the
 251  * given 'uuid', or NULL if there is no such row. */
 252 static struct ovsdb_monitor_row *
 253 ovsdb_monitor_changes_row_find(const struct ovsdb_monitor_changes *changes,
 254                                const struct uuid *uuid)
 255 {
 256     struct ovsdb_monitor_row *row;
 257
 258     HMAP_FOR_EACH_WITH_HASH (row, hmap_node, uuid_hash(uuid),
 259                              &changes->rows) {
 260         if (uuid_equals(uuid, &row->uuid)) {
 261             return row;
 262         }
 263     }
 264     return NULL;
 265 }
 266
 267 /* Allocates an array of 'mt->n_columns' ovsdb_datums and initializes them as
 268  * copies of the data in 'row' drawn from the columns represented by
 269  * mt->columns[].  Returns the array.
 270  *
 271  * If 'row' is NULL, returns NULL. */
 272 static struct ovsdb_datum *
 273 clone_monitor_row_data(const struct ovsdb_monitor_table *mt,
 274                        const struct ovsdb_row *row)
 275 {
 276     struct ovsdb_datum *data;
 277     size_t i;
 278
 279     if (!row) {
 280         return NULL;
 281     }
 282
 283     data = xmalloc(mt->n_columns * sizeof *data);
 284     for (i = 0; i < mt->n_columns; i++) {
 285         const struct ovsdb_column *c = mt->columns[i].column;
 286         const struct ovsdb_datum *src = &row->fields[c->index];
 287         struct ovsdb_datum *dst = &data[i];
 288         const struct ovsdb_type *type = &c->type;
 289
 290         ovsdb_datum_clone(dst, src, type);
 291     }
 292     return data;
 293 }
 294
 295 /* Replaces the mt->n_columns ovsdb_datums in row[] by copies of the data from
 296  * in 'row' drawn from the columns represented by mt->columns[]. */
 297 static void
 298 update_monitor_row_data(const struct ovsdb_monitor_table *mt,
 299                         const struct ovsdb_row *row,
 300                         struct ovsdb_datum *data)
 301 {
 302     size_t i;
 303
 304     for (i = 0; i < mt->n_columns; i++) {
 305         const struct ovsdb_column *c = mt->columns[i].column;
 306         const struct ovsdb_datum *src = &row->fields[c->index];
 307         struct ovsdb_datum *dst = &data[i];
 308         const struct ovsdb_type *type = &c->type;
 309
 310         if (!ovsdb_datum_equals(src, dst, type)) {
 311             ovsdb_datum_destroy(dst, type);
 312             ovsdb_datum_clone(dst, src, type);
 313         }
 314     }
 315 }
 316
 317 /* Frees all of the mt->n_columns ovsdb_datums in data[], using the types taken
 318  * from mt->columns[], plus 'data' itself. */
 319 static void
 320 free_monitor_row_data(const struct ovsdb_monitor_table *mt,
 321                       struct ovsdb_datum *data)
 322 {
 323     if (data) {
 324         size_t i;
 325
 326         for (i = 0; i < mt->n_columns; i++) {
 327             const struct ovsdb_column *c = mt->columns[i].column;
 328
 329             ovsdb_datum_destroy(&data[i], &c->type);
 330         }
 331         free(data);
 332     }
 333 }
 334
 335 /* Frees 'row', which must have been created from 'mt'. */
 336 static void
 337 ovsdb_monitor_row_destroy(const struct ovsdb_monitor_table *mt,
 338                           struct ovsdb_monitor_row *row)
 339 {
 340     if (row) {
 341         free_monitor_row_data(mt, row->old);
 342         free_monitor_row_data(mt, row->new);
 343         free(row);
 344     }
 345 }
 346
 347 static void
 348 ovsdb_monitor_columns_sort(struct ovsdb_monitor *dbmon)
 349 {
 350     int i;
 351     struct shash_node *node;
 352
 353     SHASH_FOR_EACH (node, &dbmon->tables) {
 354         struct ovsdb_monitor_table *mt = node->data;
 355
 356         qsort(mt->columns, mt->n_columns, sizeof *mt->columns,
 357               compare_ovsdb_monitor_column);
 358         for (i = 0; i < mt->n_columns; i++) {
 359             /* re-set index map due to sort */
 360             mt->columns_index_map[mt->columns[i].column->index] = i;
 361         }
 362     }
 363 }
 364
 365 void
 366 ovsdb_monitor_add_jsonrpc_monitor(struct ovsdb_monitor *dbmon,
 367                                   struct ovsdb_jsonrpc_monitor *jsonrpc_monitor)
 368 {
 369     struct jsonrpc_monitor_node *jm;
 370
 371     jm = xzalloc(sizeof *jm);
 372     jm->jsonrpc_monitor = jsonrpc_monitor;
 373     ovs_list_push_back(&dbmon->jsonrpc_monitors, &jm->node);
 374 }
 375
 376 struct ovsdb_monitor *
 377 ovsdb_monitor_create(struct ovsdb *db,
 378                      struct ovsdb_jsonrpc_monitor *jsonrpc_monitor)
 379 {
 380     struct ovsdb_monitor *dbmon;
 381
 382     dbmon = xzalloc(sizeof *dbmon);
 383
 384     ovsdb_replica_init(&dbmon->replica, &ovsdb_jsonrpc_replica_class);
 385     ovsdb_add_replica(db, &dbmon->replica);
 386     ovs_list_init(&dbmon->jsonrpc_monitors);
 387     dbmon->db = db;
 388     dbmon->n_transactions = 0;
 389     shash_init(&dbmon->tables);
 390     hmap_node_nullify(&dbmon->hmap_node);
 391     hmap_init(&dbmon->json_cache);
 392
 393     ovsdb_monitor_add_jsonrpc_monitor(dbmon, jsonrpc_monitor);
 394     return dbmon;
 395 }
 396
 397 void
 398 ovsdb_monitor_add_table(struct ovsdb_monitor *m,
 399                         const struct ovsdb_table *table)
 400 {
 401     struct ovsdb_monitor_table *mt;
 402     int i;
 403     size_t n_columns = shash_count(&table->schema->columns);
 404
 405     mt = xzalloc(sizeof *mt);
 406     mt->table = table;
 407     shash_add(&m->tables, table->schema->name, mt);
 408     hmap_init(&mt->changes);
 409     mt->columns_index_map =
 410         xmalloc(sizeof *mt->columns_index_map * n_columns);
 411     for (i = 0; i < n_columns; i++) {
 412         mt->columns_index_map[i] = -1;
 413     }
 414 }
 415
 416 const char *
 417 ovsdb_monitor_add_column(struct ovsdb_monitor *dbmon,
 418                          const struct ovsdb_table *table,
 419                          const struct ovsdb_column *column,
 420                          enum ovsdb_monitor_selection select,
 421                          bool monitored)
 422 {
 423     struct ovsdb_monitor_table *mt;
 424     struct ovsdb_monitor_column *c;
 425
 426     mt = shash_find_data(&dbmon->tables, table->schema->name);
 427
 428     /* Check for column duplication. Return duplicated column name. */
 429     if (mt->columns_index_map[column->index] != -1) {
 430         return column->name;
 431     }
 432
 433     if (mt->n_columns >= mt->allocated_columns) {
 434         mt->columns = x2nrealloc(mt->columns, &mt->allocated_columns,
 435                                  sizeof *mt->columns);
 436     }
 437
 438     mt->select |= select;
 439     mt->columns_index_map[column->index] = mt->n_columns;
 440     c = &mt->columns[mt->n_columns++];
 441     c->column = column;
 442     c->select = select;
 443     c->monitored = monitored;
 444     if (monitored) {
 445         mt->n_monitored_columns++;
 446     }
 447
 448     return NULL;
 449 }
 450
 451 static void
 452 ovsdb_monitor_condition_add_columns(struct ovsdb_monitor *dbmon,
 453                                     const struct ovsdb_table *table,
 454                                     struct ovsdb_condition *condition)
 455 {
 456     size_t n_columns;
 457     int i;
 458     const struct ovsdb_column **columns =
 459         ovsdb_condition_get_columns(condition, &n_columns);
 460
 461     for (i = 0; i < n_columns; i++) {
 462         ovsdb_monitor_add_column(dbmon, table, columns[i],
 463                                  OJMS_NONE, false);
 464     }
 465
 466     free(columns);
 467 }
 468
 469 /* Bind this session's condition to ovsdb_monitor */
 470 void
 471 ovsdb_monitor_condition_bind(struct ovsdb_monitor *dbmon,
 472                           struct ovsdb_monitor_session_condition *cond)
 473 {
 474     struct shash_node *node;
 475
 476     SHASH_FOR_EACH(node, &cond->tables) {
 477         struct ovsdb_monitor_table_condition *mtc = node->data;
 478         struct ovsdb_monitor_table *mt =
 479             shash_find_data(&dbmon->tables, mtc->table->schema->name);
 480
 481         mtc->mt = mt;
 482         ovsdb_monitor_condition_add_columns(dbmon, mtc->table,
 483                                             &mtc->new_condition);
 484     }
 485 }
 486
 487 bool
 488 ovsdb_monitor_table_exists(struct ovsdb_monitor *m,
 489                            const struct ovsdb_table *table)
 490 {
 491     return shash_find_data(&m->tables, table->schema->name);
 492 }
 493
 494 static struct ovsdb_monitor_changes *
 495 ovsdb_monitor_table_add_changes(struct ovsdb_monitor_table *mt,
 496                                 uint64_t next_txn)
 497 {
 498     struct ovsdb_monitor_changes *changes;
 499
 500     changes = xzalloc(sizeof *changes);
 501
 502     changes->transaction = next_txn;
 503     changes->mt = mt;
 504     changes->n_refs = 1;
 505     hmap_init(&changes->rows);
 506     hmap_insert(&mt->changes, &changes->hmap_node, hash_uint64(next_txn));
 507
 508     return changes;
 509 };
 510
 511 static struct ovsdb_monitor_changes *
 512 ovsdb_monitor_table_find_changes(struct ovsdb_monitor_table *mt,
 513                                  uint64_t transaction)
 514 {
 515     struct ovsdb_monitor_changes *changes;
 516     size_t hash = hash_uint64(transaction);
 517
 518     HMAP_FOR_EACH_WITH_HASH(changes, hmap_node, hash, &mt->changes) {
 519         if (changes->transaction == transaction) {
 520             return changes;
 521         }
 522     }
 523
 524     return NULL;
 525 }
 526
 527 /* Stop currently tracking changes to table 'mt' since 'transaction'. */
 528 static void
 529 ovsdb_monitor_table_untrack_changes(struct ovsdb_monitor_table *mt,
 530                                     uint64_t transaction)
 531 {
 532     struct ovsdb_monitor_changes *changes =
 533                 ovsdb_monitor_table_find_changes(mt, transaction);
 534     if (changes) {
 535         if (--changes->n_refs == 0) {
 536             hmap_remove(&mt->changes, &changes->hmap_node);
 537             ovsdb_monitor_changes_destroy(changes);
 538         }
 539     }
 540 }
 541
 542 /* Start tracking changes to table 'mt' begins from 'transaction' inclusive.
 543  */
 544 static void
 545 ovsdb_monitor_table_track_changes(struct ovsdb_monitor_table *mt,
 546                                   uint64_t transaction)
 547 {
 548     struct ovsdb_monitor_changes *changes;
 549
 550     changes = ovsdb_monitor_table_find_changes(mt, transaction);
 551     if (changes) {
 552         changes->n_refs++;
 553     } else {
 554         ovsdb_monitor_table_add_changes(mt, transaction);
 555     }
 556 }
 557
 558 static void
 559 ovsdb_monitor_changes_destroy(struct ovsdb_monitor_changes *changes)
 560 {
 561     struct ovsdb_monitor_row *row, *next;
 562
 563     HMAP_FOR_EACH_SAFE (row, next, hmap_node, &changes->rows) {
 564         hmap_remove(&changes->rows, &row->hmap_node);
 565         ovsdb_monitor_row_destroy(changes->mt, row);
 566     }
 567     hmap_destroy(&changes->rows);
 568     free(changes);
 569 }
 570
 571 static enum ovsdb_monitor_selection
 572 ovsdb_monitor_row_update_type(bool initial, const bool old, const bool new)
 573 {
 574     return initial ? OJMS_INITIAL
 575             : !old ? OJMS_INSERT
 576             : !new ? OJMS_DELETE
 577             : OJMS_MODIFY;
 578 }
 579
 580 /* Set conditional monitoring mode only if we have non-empty condition in one
 581  * of the tables at least */
 582 static inline void
 583 ovsdb_monitor_session_condition_set_mode(
 584                                   struct ovsdb_monitor_session_condition *cond)
 585 {
 586     cond->conditional = shash_count(&cond->tables) !=
 587         cond->n_true_cnd;
 588 }
 589
 590 /* Returnes an empty allocated session's condition state holder */
 591 struct ovsdb_monitor_session_condition *
 592 ovsdb_monitor_session_condition_create(void)
 593 {
 594     struct ovsdb_monitor_session_condition *condition =
 595         xzalloc(sizeof *condition);
 596
 597     condition->conditional = false;
 598     shash_init(&condition->tables);
 599     return condition;
 600 }
 601
 602 void
 603 ovsdb_monitor_session_condition_destroy(
 604                            struct ovsdb_monitor_session_condition *condition)
 605 {
 606     struct shash_node *node, *next;
 607
 608     if (!condition) {
 609         return;
 610     }
 611
 612     SHASH_FOR_EACH_SAFE (node, next, &condition->tables) {
 613         struct ovsdb_monitor_table_condition *mtc = node->data;
 614
 615         ovsdb_condition_destroy(&mtc->new_condition);
 616         ovsdb_condition_destroy(&mtc->old_condition);
 617         shash_delete(&condition->tables, node);
 618         free(mtc);
 619     }
 620     free(condition);
 621 }
 622
 623 struct ovsdb_error *
 624 ovsdb_monitor_table_condition_create(
 625                          struct ovsdb_monitor_session_condition *condition,
 626                          const struct ovsdb_table *table,
 627                          const struct json *json_cnd)
 628 {
 629     struct ovsdb_monitor_table_condition *mtc;
 630     struct ovsdb_error *error;
 631
 632     mtc = xzalloc(sizeof *mtc);
 633     mtc->table = table;
 634     ovsdb_condition_init(&mtc->old_condition);
 635     ovsdb_condition_init(&mtc->new_condition);
 636
 637     if (json_cnd) {
 638         error = ovsdb_condition_from_json(table->schema,
 639                                           json_cnd,
 640                                           NULL,
 641                                           &mtc->old_condition);
 642         if (error) {
 643             free(mtc);
 644             return error;
 645         }
 646     }
 647
 648     shash_add(&condition->tables, table->schema->name, mtc);
 649     /* On session startup old == new condition */
 650     ovsdb_condition_clone(&mtc->new_condition, &mtc->old_condition);
 651     if (ovsdb_condition_is_true(&mtc->old_condition)) {
 652         condition->n_true_cnd++;
 653         ovsdb_monitor_session_condition_set_mode(condition);
 654     }
 655
 656     return NULL;
 657 }
 658
 659 static bool
 660 ovsdb_monitor_get_table_conditions(
 661                       const struct ovsdb_monitor_table *mt,
 662                       const struct ovsdb_monitor_session_condition *condition,
 663                       struct ovsdb_condition **old_condition,
 664                       struct ovsdb_condition **new_condition)
 665 {
 666     if (!condition) {
 667         return false;
 668     }
 669
 670     struct ovsdb_monitor_table_condition *mtc =
 671         shash_find_data(&condition->tables, mt->table->schema->name);
 672
 673     if (!mtc) {
 674         return false;
 675     }
 676     *old_condition = &mtc->old_condition;
 677     *new_condition = &mtc->new_condition;
 678
 679     return true;
 680 }
 681
 682 struct ovsdb_error *
 683 ovsdb_monitor_table_condition_update(
 684                             struct ovsdb_monitor *dbmon,
 685                             struct ovsdb_monitor_session_condition *condition,
 686                             const struct ovsdb_table *table,
 687                             const struct json *cond_json)
 688 {
 689     struct ovsdb_monitor_table_condition *mtc =
 690         shash_find_data(&condition->tables, table->schema->name);
 691     struct ovsdb_error *error;
 692     struct ovsdb_condition cond = OVSDB_CONDITION_INITIALIZER(&cond);
 693
 694     if (!condition) {
 695         return NULL;
 696     }
 697
 698     error = ovsdb_condition_from_json(table->schema, cond_json,
 699                                       NULL, &cond);
 700     if (error) {
 701         return error;
 702     }
 703     ovsdb_condition_destroy(&mtc->new_condition);
 704     ovsdb_condition_clone(&mtc->new_condition, &cond);
 705     ovsdb_condition_destroy(&cond);
 706     ovsdb_monitor_condition_add_columns(dbmon,
 707                                         table,
 708                                         &mtc->new_condition);
 709
 710     return NULL;
 711 }
 712
 713 static void
 714 ovsdb_monitor_table_condition_updated(struct ovsdb_monitor_table *mt,
 715                     struct ovsdb_monitor_session_condition *condition)
 716 {
 717     struct ovsdb_monitor_table_condition *mtc =
 718         shash_find_data(&condition->tables, mt->table->schema->name);
 719
 720     if (mtc) {
 721         /* If conditional monitoring - set old condition to new condition */
 722         if (ovsdb_condition_cmp_3way(&mtc->old_condition,
 723                                      &mtc->new_condition)) {
 724             if (ovsdb_condition_is_true(&mtc->new_condition)) {
 725                                 if (!ovsdb_condition_is_true(&mtc->old_condition)) {
 726                     condition->n_true_cnd++;
 727                 }
 728             } else {
 729                 if (ovsdb_condition_is_true(&mtc->old_condition)) {
 730                     condition->n_true_cnd--;
 731                 }
 732             }
 733             ovsdb_condition_destroy(&mtc->old_condition);
 734             ovsdb_condition_clone(&mtc->old_condition, &mtc->new_condition);
 735             ovsdb_monitor_session_condition_set_mode(condition);
 736         }
 737     }
 738 }
 739
 740 static enum ovsdb_monitor_selection
 741 ovsdb_monitor_row_update_type_condition(
 742                       const struct ovsdb_monitor_table *mt,
 743                       const struct ovsdb_monitor_session_condition *condition,
 744                       bool initial,
 745                       enum ovsdb_monitor_row_type row_type,
 746                       const struct ovsdb_datum *old,
 747                       const struct ovsdb_datum *new)
 748 {
 749     struct ovsdb_condition *old_condition, *new_condition;
 750     enum ovsdb_monitor_selection type =
 751         ovsdb_monitor_row_update_type(initial, old, new);
 752
 753     if (ovsdb_monitor_get_table_conditions(mt,
 754                                            condition,
 755                                            &old_condition,
 756                                            &new_condition)) {
 757         bool old_cond = !old ? false
 758             : ovsdb_condition_empty_or_match_any(old,
 759                                                 old_condition,
 760                                                 row_type == OVSDB_MONITOR_ROW ?
 761                                                 mt->columns_index_map :
 762                                                 NULL);
 763         bool new_cond = !new ? false
 764             : ovsdb_condition_empty_or_match_any(new,
 765                                                 new_condition,
 766                                                 row_type == OVSDB_MONITOR_ROW ?
 767                                                 mt->columns_index_map :
 768                                                 NULL);
 769
 770         if (!old_cond && !new_cond) {
 771             type = OJMS_NONE;
 772         }
 773
 774         switch (type) {
 775         case OJMS_INITIAL:
 776         case OJMS_INSERT:
 777             if (!new_cond) {
 778                 type = OJMS_NONE;
 779             }
 780             break;
 781         case OJMS_MODIFY:
 782             type = !old_cond ? OJMS_INSERT : !new_cond
 783                 ? OJMS_DELETE : OJMS_MODIFY;
 784             break;
 785         case OJMS_DELETE:
 786             if (!old_cond) {
 787                 type = OJMS_NONE;
 788             }
 789             break;
 790         case OJMS_NONE:
 791             break;
 792         }
 793     }
 794     return type;
 795 }
 796
 797 static bool
 798 ovsdb_monitor_row_skip_update(const struct ovsdb_monitor_table *mt,
 799                               enum ovsdb_monitor_row_type row_type,
 800                               const struct ovsdb_datum *old,
 801                               const struct ovsdb_datum *new,
 802                               enum ovsdb_monitor_selection type,
 803                               unsigned long int *changed)
 804 {
 805     if (!(mt->select & type)) {
 806         return true;
 807     }
 808
 809     if (type == OJMS_MODIFY) {
 810         size_t i, n_changes;
 811
 812         n_changes = 0;
 813         memset(changed, 0, bitmap_n_bytes(mt->n_columns));
 814         for (i = 0; i < mt->n_columns; i++) {
 815             const struct ovsdb_column *c = mt->columns[i].column;
 816             size_t index = row_type == OVSDB_ROW ? c->index : i;
 817             if (!ovsdb_datum_equals(&old[index], &new[index], &c->type)) {
 818                 bitmap_set1(changed, i);
 819                 n_changes++;
 820             }
 821         }
 822         if (!n_changes) {
 823             /* No actual changes: presumably a row changed and then
 824              * changed back later. */
 825             return true;
 826         }
 827     }
 828
 829     return false;
 830 }
 831
 832 /* Returns JSON for a <row-update> (as described in RFC 7047) for 'row' within
 833  * 'mt', or NULL if no row update should be sent.
 834  *
 835  * The caller should specify 'initial' as true if the returned JSON is going to
 836  * be used as part of the initial reply to a "monitor" request, false if it is
 837  * going to be used as part of an "update" notification.
 838  *
 839  * 'changed' must be a scratch buffer for internal use that is at least
 840  * bitmap_n_bytes(mt->n_columns) bytes long. */
 841 static struct json *
 842 ovsdb_monitor_compose_row_update(
 843     const struct ovsdb_monitor_table *mt,
 844     const struct ovsdb_monitor_session_condition *condition OVS_UNUSED,
 845     enum ovsdb_monitor_row_type row_type OVS_UNUSED,
 846     const void *_row,
 847     bool initial, unsigned long int *changed)
 848 {
 849     const struct ovsdb_monitor_row *row = _row;
 850     enum ovsdb_monitor_selection type;
 851     struct json *old_json, *new_json;
 852     struct json *row_json;
 853     size_t i;
 854
 855     ovs_assert(row_type == OVSDB_MONITOR_ROW);
 856     type = ovsdb_monitor_row_update_type(initial, row->old, row->new);
 857     if (ovsdb_monitor_row_skip_update(mt, row_type, row->old,
 858                                       row->new, type, changed)) {
 859         return NULL;
 860     }
 861
 862     row_json = json_object_create();
 863     old_json = new_json = NULL;
 864     if (type & (OJMS_DELETE | OJMS_MODIFY)) {
 865         old_json = json_object_create();
 866         json_object_put(row_json, "old", old_json);
 867     }
 868     if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
 869         new_json = json_object_create();
 870         json_object_put(row_json, "new", new_json);
 871     }
 872     for (i = 0; i < mt->n_monitored_columns; i++) {
 873         const struct ovsdb_monitor_column *c = &mt->columns[i];
 874
 875         if (!c->monitored || !(type & c->select))  {
 876             /* We don't care about this type of change for this
 877              * particular column (but we will care about it for some
 878              * other column). */
 879             continue;
 880         }
 881
 882         if ((type == OJMS_MODIFY && bitmap_is_set(changed, i))
 883             || type == OJMS_DELETE) {
 884             json_object_put(old_json, c->column->name,
 885                             ovsdb_datum_to_json(&row->old[i],
 886                                                 &c->column->type));
 887         }
 888         if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
 889             json_object_put(new_json, c->column->name,
 890                             ovsdb_datum_to_json(&row->new[i],
 891                                                 &c->column->type));
 892         }
 893     }
 894
 895     return row_json;
 896 }
 897
 898 /* Returns JSON for a <row-update2> (as described in ovsdb-server(1) mapage)
 899  * for 'row' within * 'mt', or NULL if no row update should be sent.
 900  *
 901  * The caller should specify 'initial' as true if the returned JSON is
 902  * going to be used as part of the initial reply to a "monitor_cond" request,
 903  * false if it is going to be used as part of an "update2" notification.
 904  *
 905  * 'changed' must be a scratch buffer for internal use that is at least
 906  * bitmap_n_bytes(mt->n_columns) bytes long. */
 907 static struct json *
 908 ovsdb_monitor_compose_row_update2(
 909     const struct ovsdb_monitor_table *mt,
 910     const struct ovsdb_monitor_session_condition *condition,
 911     enum ovsdb_monitor_row_type row_type,
 912     const void *_row,
 913     bool initial, unsigned long int *changed)
 914 {
 915     enum ovsdb_monitor_selection type;
 916     struct json *row_update2, *diff_json;
 917     const struct ovsdb_datum *old, *new;
 918     size_t i;
 919
 920     if (row_type == OVSDB_MONITOR_ROW) {
 921         old = ((const struct ovsdb_monitor_row *)_row)->old;;
 922         new = ((const struct ovsdb_monitor_row *)_row)->new;
 923     } else {
 924         old = new = ((const struct ovsdb_row *)_row)->fields;
 925     }
 926
 927     type = ovsdb_monitor_row_update_type_condition(mt, condition, initial,
 928                                                    row_type, old, new);
 929     if (ovsdb_monitor_row_skip_update(mt, row_type, old, new, type, changed)) {
 930         return NULL;
 931     }
 932
 933     row_update2 = json_object_create();
 934     if (type == OJMS_DELETE) {
 935         json_object_put(row_update2, "delete", json_null_create());
 936     } else {
 937         diff_json = json_object_create();
 938         const char *op;
 939
 940         for (i = 0; i < mt->n_monitored_columns; i++) {
 941             const struct ovsdb_monitor_column *c = &mt->columns[i];
 942             size_t index = row_type == OVSDB_ROW ? c->column->index : i;
 943             if (!c->monitored || !(type & c->select))  {
 944                 /* We don't care about this type of change for this
 945                  * particular column (but we will care about it for some
 946                  * other column). */
 947                 continue;
 948             }
 949
 950             if (type == OJMS_MODIFY) {
 951                 struct ovsdb_datum diff;
 952
 953                 if (!bitmap_is_set(changed, i)) {
 954                     continue;
 955                 }
 956
 957                 ovsdb_datum_diff(&diff ,&old[index], &new[index],
 958                                         &c->column->type);
 959                 json_object_put(diff_json, c->column->name,
 960                                 ovsdb_datum_to_json(&diff, &c->column->type));
 961                 ovsdb_datum_destroy(&diff, &c->column->type);
 962             } else {
 963                 if (!ovsdb_datum_is_default(&new[index], &c->column->type)) {
 964                     json_object_put(diff_json, c->column->name,
 965                                     ovsdb_datum_to_json(&new[index],
 966                                                         &c->column->type));
 967                 }
 968             }
 969         }
 970
 971         op = type == OJMS_INITIAL ? "initial"
 972                                   : type == OJMS_MODIFY ? "modify" : "insert";
 973         json_object_put(row_update2, op, diff_json);
 974     }
 975
 976     return row_update2;
 977 }
 978
 979 static size_t
 980 ovsdb_monitor_max_columns(struct ovsdb_monitor *dbmon)
 981 {
 982     struct shash_node *node;
 983     size_t max_columns = 0;
 984
 985     SHASH_FOR_EACH (node, &dbmon->tables) {
 986         struct ovsdb_monitor_table *mt = node->data;
 987
 988         max_columns = MAX(max_columns, mt->n_columns);
 989     }
 990
 991     return max_columns;
 992 }
 993
 994 static void
 995 ovsdb_monitor_add_json_row(struct json **json, const char *table_name,
 996                            struct json **table_json, struct json *row_json,
 997                            const struct uuid *row_uuid)
 998 {
 999     char uuid[UUID_LEN + 1];
1000
1001     /* Create JSON object for transaction overall. */
1002     if (!*json) {
1003         *json = json_object_create();
1004     }
1005
1006     /* Create JSON object for transaction on this table. */
1007     if (!*table_json) {
1008         *table_json = json_object_create();
1009         json_object_put(*json, table_name, *table_json);
1010     }
1011
1012     /* Add JSON row to JSON table. */
1013     snprintf(uuid, sizeof uuid, UUID_FMT, UUID_ARGS(row_uuid));
1014     json_object_put(*table_json, uuid, row_json);
1015 }
1016
1017 /* Constructs and returns JSON for a <table-updates> object (as described in
1018  * RFC 7047) for all the outstanding changes within 'monitor', starting from
1019  * 'transaction'.  */
1020 static struct json*
1021 ovsdb_monitor_compose_update(
1022                       struct ovsdb_monitor *dbmon,
1023                       bool initial, uint64_t transaction,
1024                       const struct ovsdb_monitor_session_condition *condition,
1025                       compose_row_update_cb_func row_update)
1026 {
1027     struct shash_node *node;
1028     struct json *json;
1029     size_t max_columns = ovsdb_monitor_max_columns(dbmon);
1030     unsigned long int *changed = xmalloc(bitmap_n_bytes(max_columns));
1031
1032     json = NULL;
1033     SHASH_FOR_EACH (node, &dbmon->tables) {
1034         struct ovsdb_monitor_table *mt = node->data;
1035         struct ovsdb_monitor_row *row, *next;
1036         struct ovsdb_monitor_changes *changes;
1037         struct json *table_json = NULL;
1038
1039         changes = ovsdb_monitor_table_find_changes(mt, transaction);
1040         if (!changes) {
1041             continue;
1042         }
1043
1044                 HMAP_FOR_EACH_SAFE (row, next, hmap_node, &changes->rows) {
1045                         struct json *row_json;
1046                         row_json = (*row_update)(mt, condition, OVSDB_MONITOR_ROW, row,
1047                                                                          initial, changed);
1048                         if (row_json) {
1049                                 ovsdb_monitor_add_json_row(&json, mt->table->schema->name,
1050                                                                                    &table_json, row_json,
1051                                                                                    &row->uuid);
1052                         }
1053                 }
1054         }
1055     free(changed);
1056
1057     return json;
1058 }
1059
1060 static struct json*
1061 ovsdb_monitor_compose_cond_change_update(
1062                     struct ovsdb_monitor *dbmon,
1063                     struct ovsdb_monitor_session_condition *condition)
1064 {
1065     struct shash_node *node;
1066     struct json *json = NULL;
1067     size_t max_columns = ovsdb_monitor_max_columns(dbmon);
1068     unsigned long int *changed = xmalloc(bitmap_n_bytes(max_columns));
1069
1070     SHASH_FOR_EACH (node, &dbmon->tables) {
1071         struct ovsdb_monitor_table *mt = node->data;
1072         struct ovsdb_row *row;
1073         struct json *table_json = NULL;
1074         struct ovsdb_condition *old_condition, *new_condition;
1075
1076         if (!ovsdb_monitor_get_table_conditions(mt,
1077                                                 condition,
1078                                                 &old_condition,
1079                                                 &new_condition) ||
1080             !ovsdb_condition_cmp_3way(old_condition, new_condition)) {
1081             /* Nothing to update on this table */
1082             continue;
1083         }
1084
1085         /* Iterate over all rows in table */
1086         HMAP_FOR_EACH (row, hmap_node, &mt->table->rows) {
1087             struct json *row_json;
1088
1089             row_json = ovsdb_monitor_compose_row_update2(mt, condition,
1090                                                          OVSDB_ROW, row,
1091                                                          false, changed);
1092             if (row_json) {
1093                 ovsdb_monitor_add_json_row(&json, mt->table->schema->name,
1094                                            &table_json, row_json,
1095                                            ovsdb_row_get_uuid(row));
1096             }
1097         }
1098         ovsdb_monitor_table_condition_updated(mt, condition);
1099     }
1100     free(changed);
1101
1102     return json;
1103 }
1104
1105 /* Returns JSON for a <table-updates> object (as described in RFC 7047)
1106  * for all the outstanding changes within 'monitor' that starts from
1107  * '*unflushed'.
1108  * If cond_updated is true all rows in the db that match conditions will be
1109  * sent.
1110  *
1111  * The caller should specify 'initial' as true if the returned JSON is going to
1112  * be used as part of the initial reply to a "monitor" request, false if it is
1113  * going to be used as part of an "update" notification. */
1114 struct json *
1115 ovsdb_monitor_get_update(
1116              struct ovsdb_monitor *dbmon,
1117              bool initial, bool cond_updated,
1118              uint64_t *unflushed_,
1119              struct ovsdb_monitor_session_condition *condition,
1120              enum ovsdb_monitor_version version)
1121 {
1122     struct ovsdb_monitor_json_cache_node *cache_node = NULL;
1123     struct shash_node *node;
1124     struct json *json;
1125     const uint64_t unflushed = *unflushed_;
1126     const uint64_t next_unflushed = dbmon->n_transactions + 1;
1127
1128     ovs_assert(cond_updated ? unflushed == next_unflushed : true);
1129
1130     /* Return a clone of cached json if one exists. Otherwise,
1131      * generate a new one and add it to the cache.  */
1132     if (!condition || (!condition->conditional && !cond_updated)) {
1133         cache_node = ovsdb_monitor_json_cache_search(dbmon, version,
1134                                                      unflushed);
1135     }
1136     if (cache_node) {
1137         json = cache_node->json ? json_clone(cache_node->json) : NULL;
1138     } else {
1139         if (version == OVSDB_MONITOR_V1) {
1140             json =
1141                ovsdb_monitor_compose_update(dbmon, initial, unflushed,
1142                                             condition,
1143                                             ovsdb_monitor_compose_row_update);
1144         } else {
1145             ovs_assert(version == OVSDB_MONITOR_V2);
1146             if (!cond_updated) {
1147                                 json = ovsdb_monitor_compose_update(dbmon, initial, unflushed,
1148                                                                                         condition,
1149                                                                                         ovsdb_monitor_compose_row_update2);
1150
1151                                 if (!condition || !condition->conditional) {
1152                                         ovsdb_monitor_json_cache_insert(dbmon, version, unflushed,
1153                                                                                                         json);
1154                                 }
1155                         } else {
1156                 /* Compose update on whole db due to condition update.
1157                    Session must be flushed (change list is empty)*/
1158                                 json =
1159                                         ovsdb_monitor_compose_cond_change_update(dbmon, condition);
1160                         }
1161                 }
1162     }
1163
1164     /* Maintain transaction id of 'changes'. */
1165     SHASH_FOR_EACH (node, &dbmon->tables) {
1166         struct ovsdb_monitor_table *mt = node->data;
1167
1168         ovsdb_monitor_table_untrack_changes(mt, unflushed);
1169         ovsdb_monitor_table_track_changes(mt, next_unflushed);
1170     }
1171     *unflushed_ = next_unflushed;
1172
1173     return json;
1174 }
1175
1176 bool
1177 ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon,
1178                           uint64_t next_transaction)
1179 {
1180     ovs_assert(next_transaction <= dbmon->n_transactions + 1);
1181     return (next_transaction <= dbmon->n_transactions);
1182 }
1183
1184 void
1185 ovsdb_monitor_table_add_select(struct ovsdb_monitor *dbmon,
1186                                const struct ovsdb_table *table,
1187                                enum ovsdb_monitor_selection select)
1188 {
1189     struct ovsdb_monitor_table * mt;
1190
1191     mt = shash_find_data(&dbmon->tables, table->schema->name);
1192     mt->select |= select;
1193 }
1194
1195  /*
1196  * If a row's change type (insert, delete or modify) matches that of
1197  * the monitor, they should be sent to the monitor's clients as updates.
1198  * Of cause, the monitor should also internally update with this change.
1199  *
1200  * When a change type does not require client side update, the monitor
1201  * may still need to keep track of certain changes in order to generate
1202  * correct future updates.  For example, the monitor internal state should
1203  * be updated whenever a new row is inserted, in order to generate the
1204  * correct initial state, regardless if a insert change type is being
1205  * monitored.
1206  *
1207  * On the other hand, if a transaction only contains changes to columns
1208  * that are not monitored, this transaction can be safely ignored by the
1209  * monitor.
1210  *
1211  * Thus, the order of the declaration is important:
1212  * 'OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE' always implies
1213  * 'OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE', but not vice versa.  */
1214 enum ovsdb_monitor_changes_efficacy {
1215     OVSDB_CHANGES_NO_EFFECT,                /* Monitor does not care about this
1216                                                change.  */
1217     OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE,  /* Monitor internal updates. */
1218     OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE,  /* Client needs to be updated.  */
1219 };
1220
1221 struct ovsdb_monitor_aux {
1222     const struct ovsdb_monitor *monitor;
1223     struct ovsdb_monitor_table *mt;
1224     enum ovsdb_monitor_changes_efficacy efficacy;
1225 };
1226
1227 static void
1228 ovsdb_monitor_init_aux(struct ovsdb_monitor_aux *aux,
1229                        const struct ovsdb_monitor *m)
1230 {
1231     aux->monitor = m;
1232     aux->mt = NULL;
1233     aux->efficacy = OVSDB_CHANGES_NO_EFFECT;
1234 }
1235
1236 static void
1237 ovsdb_monitor_changes_update(const struct ovsdb_row *old,
1238                              const struct ovsdb_row *new,
1239                              const struct ovsdb_monitor_table *mt,
1240                              struct ovsdb_monitor_changes *changes)
1241 {
1242     const struct uuid *uuid = ovsdb_row_get_uuid(new ? new : old);
1243     struct ovsdb_monitor_row *change;
1244
1245     change = ovsdb_monitor_changes_row_find(changes, uuid);
1246     if (!change) {
1247         change = xzalloc(sizeof *change);
1248         hmap_insert(&changes->rows, &change->hmap_node, uuid_hash(uuid));
1249         change->uuid = *uuid;
1250         change->old = clone_monitor_row_data(mt, old);
1251         change->new = clone_monitor_row_data(mt, new);
1252     } else {
1253         if (new) {
1254             update_monitor_row_data(mt, new, change->new);
1255         } else {
1256             free_monitor_row_data(mt, change->new);
1257             change->new = NULL;
1258
1259             if (!change->old) {
1260                 /* This row was added then deleted.  Forget about it. */
1261                 hmap_remove(&changes->rows, &change->hmap_node);
1262                 free(change);
1263             }
1264         }
1265     }
1266 }
1267
1268 static bool
1269 ovsdb_monitor_columns_changed(const struct ovsdb_monitor_table *mt,
1270                               const unsigned long int *changed)
1271 {
1272     size_t i;
1273
1274     for (i = 0; i < mt->n_columns; i++) {
1275         size_t column_index = mt->columns[i].column->index;
1276
1277         if (bitmap_is_set(changed, column_index)) {
1278             return true;
1279         }
1280     }
1281
1282     return false;
1283 }
1284
1285 /* Return the efficacy of a row's change to a monitor table.
1286  *
1287  * Please see the block comment above 'ovsdb_monitor_changes_efficacy'
1288  * definition form more information.  */
1289 static enum ovsdb_monitor_changes_efficacy
1290 ovsdb_monitor_changes_classify(enum ovsdb_monitor_selection type,
1291                                const struct ovsdb_monitor_table *mt,
1292                                const unsigned long int *changed)
1293 {
1294     if (type == OJMS_MODIFY &&
1295         !ovsdb_monitor_columns_changed(mt, changed)) {
1296         return OVSDB_CHANGES_NO_EFFECT;
1297     }
1298
1299     if (type == OJMS_MODIFY) {
1300         /* Condition might turn a modify operation to insert or delete */
1301         type |= OJMS_INSERT | OJMS_DELETE;
1302     }
1303
1304     return (mt->select & type)
1305                 ?  OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE
1306                 :  OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE;
1307 }
1308
1309 static bool
1310 ovsdb_monitor_change_cb(const struct ovsdb_row *old,
1311                         const struct ovsdb_row *new,
1312                         const unsigned long int *changed,
1313                         void *aux_)
1314 {
1315     struct ovsdb_monitor_aux *aux = aux_;
1316     const struct ovsdb_monitor *m = aux->monitor;
1317     struct ovsdb_table *table = new ? new->table : old->table;
1318     struct ovsdb_monitor_table *mt;
1319     struct ovsdb_monitor_changes *changes;
1320
1321     if (!aux->mt || table != aux->mt->table) {
1322         aux->mt = shash_find_data(&m->tables, table->schema->name);
1323         if (!aux->mt) {
1324             /* We don't care about rows in this table at all.  Tell the caller
1325              * to skip it.  */
1326             return false;
1327         }
1328     }
1329     mt = aux->mt;
1330
1331     enum ovsdb_monitor_selection type =
1332         ovsdb_monitor_row_update_type(false, old, new);
1333     enum ovsdb_monitor_changes_efficacy efficacy =
1334         ovsdb_monitor_changes_classify(type, mt, changed);
1335
1336     HMAP_FOR_EACH(changes, hmap_node, &mt->changes) {
1337         if (efficacy > OVSDB_CHANGES_NO_EFFECT) {
1338             ovsdb_monitor_changes_update(old, new, mt, changes);
1339         }
1340     }
1341     if (aux->efficacy < efficacy) {
1342         aux->efficacy = efficacy;
1343     }
1344
1345     return true;
1346 }
1347
1348 void
1349 ovsdb_monitor_get_initial(const struct ovsdb_monitor *dbmon)
1350 {
1351     struct shash_node *node;
1352
1353     SHASH_FOR_EACH (node, &dbmon->tables) {
1354         struct ovsdb_monitor_table *mt = node->data;
1355
1356         if (mt->select & OJMS_INITIAL) {
1357             struct ovsdb_row *row;
1358             struct ovsdb_monitor_changes *changes;
1359
1360             changes = ovsdb_monitor_table_find_changes(mt, 0);
1361             if (!changes) {
1362                 changes = ovsdb_monitor_table_add_changes(mt, 0);
1363                 HMAP_FOR_EACH (row, hmap_node, &mt->table->rows) {
1364                     ovsdb_monitor_changes_update(NULL, row, mt, changes);
1365                 }
1366             } else {
1367                 changes->n_refs++;
1368             }
1369         }
1370     }
1371 }
1372
1373 void
1374 ovsdb_monitor_remove_jsonrpc_monitor(struct ovsdb_monitor *dbmon,
1375                    struct ovsdb_jsonrpc_monitor *jsonrpc_monitor,
1376                    uint64_t unflushed)
1377 {
1378     struct jsonrpc_monitor_node *jm;
1379
1380     if (ovs_list_is_empty(&dbmon->jsonrpc_monitors)) {
1381         ovsdb_monitor_destroy(dbmon);
1382         return;
1383     }
1384
1385     /* Find and remove the jsonrpc monitor from the list.  */
1386     LIST_FOR_EACH(jm, node, &dbmon->jsonrpc_monitors) {
1387         if (jm->jsonrpc_monitor == jsonrpc_monitor) {
1388             /* Release the tracked changes. */
1389             struct shash_node *node;
1390             SHASH_FOR_EACH (node, &dbmon->tables) {
1391                 struct ovsdb_monitor_table *mt = node->data;
1392                 ovsdb_monitor_table_untrack_changes(mt, unflushed);
1393             }
1394             ovs_list_remove(&jm->node);
1395             free(jm);
1396
1397             /* Destroy ovsdb monitor if this is the last user.  */
1398             if (ovs_list_is_empty(&dbmon->jsonrpc_monitors)) {
1399                 ovsdb_monitor_destroy(dbmon);
1400             }
1401
1402             return;
1403         };
1404     }
1405
1406     /* Should never reach here. jsonrpc_monitor should be on the list.  */
1407     OVS_NOT_REACHED();
1408 }
1409
1410 static bool
1411 ovsdb_monitor_table_equal(const struct ovsdb_monitor_table *a,
1412                           const struct ovsdb_monitor_table *b)
1413 {
1414     size_t i;
1415
1416     ovs_assert(b->n_columns == b->n_monitored_columns);
1417
1418     if ((a->table != b->table) ||
1419         (a->select != b->select) ||
1420         (a->n_monitored_columns != b->n_monitored_columns)) {
1421         return false;
1422     }
1423
1424     /* Compare only monitored columns that must be sorted already */
1425     for (i = 0; i < a->n_monitored_columns; i++) {
1426         if ((a->columns[i].column != b->columns[i].column) ||
1427             (a->columns[i].select != b->columns[i].select)) {
1428             return false;
1429         }
1430     }
1431     return true;
1432 }
1433
1434 static bool
1435 ovsdb_monitor_equal(const struct ovsdb_monitor *a,
1436                     const struct ovsdb_monitor *b)
1437 {
1438     struct shash_node *node;
1439
1440     if (shash_count(&a->tables) != shash_count(&b->tables)) {
1441         return false;
1442     }
1443
1444     SHASH_FOR_EACH(node, &a->tables) {
1445         const struct ovsdb_monitor_table *mta = node->data;
1446         const struct ovsdb_monitor_table *mtb;
1447
1448         mtb = shash_find_data(&b->tables, node->name);
1449         if (!mtb) {
1450             return false;
1451         }
1452
1453         if (!ovsdb_monitor_table_equal(mta, mtb)) {
1454             return false;
1455         }
1456     }
1457
1458     return true;
1459 }
1460
1461 static size_t
1462 ovsdb_monitor_hash(const struct ovsdb_monitor *dbmon, size_t basis)
1463 {
1464     const struct shash_node **nodes;
1465     size_t i, j, n;
1466
1467     nodes = shash_sort(&dbmon->tables);
1468     n = shash_count(&dbmon->tables);
1469
1470     for (i = 0; i < n; i++) {
1471         struct ovsdb_monitor_table *mt = nodes[i]->data;
1472
1473         basis = hash_pointer(mt->table, basis);
1474         basis = hash_3words(mt->select, mt->n_columns, basis);
1475
1476         for (j = 0; j < mt->n_columns; j++) {
1477             basis = hash_pointer(mt->columns[j].column, basis);
1478             basis = hash_2words(mt->columns[j].select, basis);
1479         }
1480     }
1481     free(nodes);
1482
1483     return basis;
1484 }
1485
1486 struct ovsdb_monitor *
1487 ovsdb_monitor_add(struct ovsdb_monitor *new_dbmon)
1488 {
1489     struct ovsdb_monitor *dbmon;
1490     size_t hash;
1491
1492     /* New_dbmon should be associated with only one jsonrpc
1493      * connections.  */
1494     ovs_assert(ovs_list_is_singleton(&new_dbmon->jsonrpc_monitors));
1495
1496     ovsdb_monitor_columns_sort(new_dbmon);
1497
1498     hash = ovsdb_monitor_hash(new_dbmon, 0);
1499     HMAP_FOR_EACH_WITH_HASH(dbmon, hmap_node, hash, &ovsdb_monitors) {
1500         if (ovsdb_monitor_equal(dbmon,  new_dbmon)) {
1501             return dbmon;
1502         }
1503     }
1504
1505     hmap_insert(&ovsdb_monitors, &new_dbmon->hmap_node, hash);
1506     return new_dbmon;
1507 }
1508
1509 static void
1510 ovsdb_monitor_destroy(struct ovsdb_monitor *dbmon)
1511 {
1512     struct shash_node *node;
1513
1514     ovs_list_remove(&dbmon->replica.node);
1515
1516     if (!hmap_node_is_null(&dbmon->hmap_node)) {
1517         hmap_remove(&ovsdb_monitors, &dbmon->hmap_node);
1518     }
1519
1520     ovsdb_monitor_json_cache_flush(dbmon);
1521     hmap_destroy(&dbmon->json_cache);
1522
1523     SHASH_FOR_EACH (node, &dbmon->tables) {
1524         struct ovsdb_monitor_table *mt = node->data;
1525         struct ovsdb_monitor_changes *changes, *next;
1526
1527         HMAP_FOR_EACH_SAFE (changes, next, hmap_node, &mt->changes) {
1528             hmap_remove(&mt->changes, &changes->hmap_node);
1529             ovsdb_monitor_changes_destroy(changes);
1530         }
1531         hmap_destroy(&mt->changes);
1532         free(mt->columns);
1533         free(mt->columns_index_map);
1534         free(mt);
1535     }
1536     shash_destroy(&dbmon->tables);
1537     free(dbmon);
1538 }
1539
1540 static struct ovsdb_error *
1541 ovsdb_monitor_commit(struct ovsdb_replica *replica,
1542                      const struct ovsdb_txn *txn,
1543                      bool durable OVS_UNUSED)
1544 {
1545     struct ovsdb_monitor *m = ovsdb_monitor_cast(replica);
1546     struct ovsdb_monitor_aux aux;
1547
1548     ovsdb_monitor_init_aux(&aux, m);
1549     /* Update ovsdb_monitor's transaction number for
1550      * each transaction, before calling ovsdb_monitor_change_cb().  */
1551     m->n_transactions++;
1552     ovsdb_txn_for_each_change(txn, ovsdb_monitor_change_cb, &aux);
1553
1554     switch(aux.efficacy) {
1555     case OVSDB_CHANGES_NO_EFFECT:
1556         /* The transaction is ignored by the monitor.
1557          * Roll back the 'n_transactions' as if the transaction
1558          * has never happened. */
1559         m->n_transactions--;
1560         break;
1561     case OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE:
1562         /* Nothing.  */
1563         break;
1564     case  OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE:
1565         ovsdb_monitor_json_cache_flush(m);
1566         break;
1567     }
1568
1569     return NULL;
1570 }
1571
1572 static void
1573 ovsdb_monitor_destroy_callback(struct ovsdb_replica *replica)
1574 {
1575     struct ovsdb_monitor *dbmon = ovsdb_monitor_cast(replica);
1576     struct jsonrpc_monitor_node *jm, *next;
1577
1578     /* Delete all front end monitors. Removing the last front
1579      * end monitor will also destroy the corresponding 'ovsdb_monitor'.
1580      * ovsdb monitor will also be destroied.  */
1581     LIST_FOR_EACH_SAFE(jm, next, node, &dbmon->jsonrpc_monitors) {
1582         ovsdb_jsonrpc_monitor_destroy(jm->jsonrpc_monitor);
1583     }
1584 }
1585
1586 /* Add some memory usage statics for monitors into 'usage', for use with
1587  * memory_report().  */
1588 void
1589 ovsdb_monitor_get_memory_usage(struct simap *usage)
1590 {
1591     struct ovsdb_monitor *dbmon;
1592     simap_put(usage, "monitors", hmap_count(&ovsdb_monitors));
1593
1594     HMAP_FOR_EACH(dbmon, hmap_node,  &ovsdb_monitors) {
1595         simap_increase(usage, "json-caches", hmap_count(&dbmon->json_cache));
1596     }
1597 }
1598
1599 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class = {
1600     ovsdb_monitor_commit,
1601     ovsdb_monitor_destroy_callback,
1602 };