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 "dynamic-string.h"
  24 #include "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 "simap.h"
  31 #include "hash.h"
  32 #include "table.h"
  33 #include "hash.h"
  34 #include "timeval.h"
  35 #include "transaction.h"
  36 #include "jsonrpc-server.h"
  37 #include "monitor.h"
  38 #include "openvswitch/vlog.h"
  39
  40
  41 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class;
  42 static struct hmap ovsdb_monitors = HMAP_INITIALIZER(&ovsdb_monitors);
  43
  44 /*  Backend monitor.
  45  *
  46  *  ovsdb_monitor keep track of the ovsdb changes.
  47  */
  48
  49 /* A collection of tables being monitored. */
  50 struct ovsdb_monitor {
  51     struct ovsdb_replica replica;
  52     struct shash tables;     /* Holds "struct ovsdb_monitor_table"s. */
  53     struct ovs_list jsonrpc_monitors;  /* Contains "jsonrpc_monitor_node"s. */
  54     struct ovsdb *db;
  55     uint64_t n_transactions;      /* Count number of committed transactions. */
  56     struct hmap_node hmap_node;   /* Elements within ovsdb_monitors.  */
  57     struct hmap json_cache;       /* Contains "ovsdb_monitor_json_cache_node"s.*/
  58 };
  59
  60 /* A json object of updates between 'from_txn' and 'dbmon->n_transactions'
  61  * inclusive.  */
  62 struct ovsdb_monitor_json_cache_node {
  63     struct hmap_node hmap_node;   /* Elements in json cache. */
  64     uint64_t from_txn;
  65     struct json *json;            /* Null, or a cloned of json */
  66 };
  67
  68 struct jsonrpc_monitor_node {
  69     struct ovsdb_jsonrpc_monitor *jsonrpc_monitor;
  70     struct ovs_list node;
  71 };
  72
  73 /* A particular column being monitored. */
  74 struct ovsdb_monitor_column {
  75     const struct ovsdb_column *column;
  76     enum ovsdb_monitor_selection select;
  77 };
  78
  79 /* A row that has changed in a monitored table. */
  80 struct ovsdb_monitor_row {
  81     struct hmap_node hmap_node; /* In ovsdb_jsonrpc_monitor_table.changes. */
  82     struct uuid uuid;           /* UUID of row that changed. */
  83     struct ovsdb_datum *old;    /* Old data, NULL for an inserted row. */
  84     struct ovsdb_datum *new;    /* New data, NULL for a deleted row. */
  85 };
  86
  87 /* Contains 'struct ovsdb_monitor_row's for rows that have been
  88  * updated but not yet flushed to all the jsonrpc connection.
  89  *
  90  * 'n_refs' represent the number of jsonrpc connections that have
  91  * not received updates. Generate the update for the last jsonprc
  92  * connection will also destroy the whole "struct ovsdb_monitor_changes"
  93  * object.
  94  *
  95  * 'transaction' stores the first update's transaction id.
  96  * */
  97 struct ovsdb_monitor_changes {
  98     struct ovsdb_monitor_table *mt;
  99     struct hmap rows;
 100     int n_refs;
 101     uint64_t transaction;
 102     struct hmap_node hmap_node;  /* Element in ovsdb_monitor_tables' changes
 103                                     hmap.  */
 104 };
 105
 106 /* A particular table being monitored. */
 107 struct ovsdb_monitor_table {
 108     const struct ovsdb_table *table;
 109
 110     /* This is the union (bitwise-OR) of the 'select' values in all of the
 111      * members of 'columns' below. */
 112     enum ovsdb_monitor_selection select;
 113
 114     /* Columns being monitored. */
 115     struct ovsdb_monitor_column *columns;
 116     size_t n_columns;
 117
 118     /* Contains 'ovsdb_monitor_changes' indexed by 'transaction'. */
 119     struct hmap changes;
 120 };
 121
 122 typedef struct json *
 123 (*compose_row_update_cb_func)(const struct ovsdb_monitor_table *mt,
 124                               const struct ovsdb_monitor_row *row,
 125                               bool initial, unsigned long int *changed);
 126
 127 static void ovsdb_monitor_destroy(struct ovsdb_monitor *dbmon);
 128 static struct ovsdb_monitor_changes * ovsdb_monitor_table_add_changes(
 129     struct ovsdb_monitor_table *mt, uint64_t next_txn);
 130 static struct ovsdb_monitor_changes *ovsdb_monitor_table_find_changes(
 131     struct ovsdb_monitor_table *mt, uint64_t unflushed);
 132 static void ovsdb_monitor_changes_destroy(
 133                                   struct ovsdb_monitor_changes *changes);
 134 static void ovsdb_monitor_table_track_changes(struct ovsdb_monitor_table *mt,
 135                                   uint64_t unflushed);
 136
 137 static struct ovsdb_monitor_json_cache_node *
 138 ovsdb_monitor_json_cache_search(const struct ovsdb_monitor *dbmon,
 139                                 uint64_t from_txn)
 140 {
 141     struct ovsdb_monitor_json_cache_node *node;
 142     uint32_t hash = hash_uint64(from_txn);
 143
 144     HMAP_FOR_EACH_WITH_HASH(node, hmap_node, hash, &dbmon->json_cache) {
 145         if (node->from_txn == from_txn) {
 146             return node;
 147         }
 148     }
 149
 150     return NULL;
 151 }
 152
 153 static void
 154 ovsdb_monitor_json_cache_insert(struct ovsdb_monitor *dbmon,
 155                                 uint64_t from_txn, struct json *json)
 156 {
 157     struct ovsdb_monitor_json_cache_node *node;
 158     uint32_t hash;
 159
 160     node = xmalloc(sizeof *node);
 161
 162     hash = hash_uint64(from_txn);
 163     node->from_txn = from_txn;
 164     node->json = json ? json_clone(json) : NULL;
 165
 166     hmap_insert(&dbmon->json_cache, &node->hmap_node, hash);
 167 }
 168
 169 static void
 170 ovsdb_monitor_json_cache_flush(struct ovsdb_monitor *dbmon)
 171 {
 172     struct ovsdb_monitor_json_cache_node *node, *next;
 173
 174     HMAP_FOR_EACH_SAFE(node, next, hmap_node, &dbmon->json_cache) {
 175         hmap_remove(&dbmon->json_cache, &node->hmap_node);
 176         json_destroy(node->json);
 177         free(node);
 178     }
 179 }
 180
 181 static int
 182 compare_ovsdb_monitor_column(const void *a_, const void *b_)
 183 {
 184     const struct ovsdb_monitor_column *a = a_;
 185     const struct ovsdb_monitor_column *b = b_;
 186
 187     return a->column < b->column ? -1 : a->column > b->column;
 188 }
 189
 190 static struct ovsdb_monitor *
 191 ovsdb_monitor_cast(struct ovsdb_replica *replica)
 192 {
 193     ovs_assert(replica->class == &ovsdb_jsonrpc_replica_class);
 194     return CONTAINER_OF(replica, struct ovsdb_monitor, replica);
 195 }
 196
 197 /* Finds and returns the ovsdb_monitor_row in 'mt->changes->rows' for the
 198  * given 'uuid', or NULL if there is no such row. */
 199 static struct ovsdb_monitor_row *
 200 ovsdb_monitor_changes_row_find(const struct ovsdb_monitor_changes *changes,
 201                                const struct uuid *uuid)
 202 {
 203     struct ovsdb_monitor_row *row;
 204
 205     HMAP_FOR_EACH_WITH_HASH (row, hmap_node, uuid_hash(uuid),
 206                              &changes->rows) {
 207         if (uuid_equals(uuid, &row->uuid)) {
 208             return row;
 209         }
 210     }
 211     return NULL;
 212 }
 213
 214 /* Allocates an array of 'mt->n_columns' ovsdb_datums and initializes them as
 215  * copies of the data in 'row' drawn from the columns represented by
 216  * mt->columns[].  Returns the array.
 217  *
 218  * If 'row' is NULL, returns NULL. */
 219 static struct ovsdb_datum *
 220 clone_monitor_row_data(const struct ovsdb_monitor_table *mt,
 221                        const struct ovsdb_row *row)
 222 {
 223     struct ovsdb_datum *data;
 224     size_t i;
 225
 226     if (!row) {
 227         return NULL;
 228     }
 229
 230     data = xmalloc(mt->n_columns * sizeof *data);
 231     for (i = 0; i < mt->n_columns; i++) {
 232         const struct ovsdb_column *c = mt->columns[i].column;
 233         const struct ovsdb_datum *src = &row->fields[c->index];
 234         struct ovsdb_datum *dst = &data[i];
 235         const struct ovsdb_type *type = &c->type;
 236
 237         ovsdb_datum_clone(dst, src, type);
 238     }
 239     return data;
 240 }
 241
 242 /* Replaces the mt->n_columns ovsdb_datums in row[] by copies of the data from
 243  * in 'row' drawn from the columns represented by mt->columns[]. */
 244 static void
 245 update_monitor_row_data(const struct ovsdb_monitor_table *mt,
 246                         const struct ovsdb_row *row,
 247                         struct ovsdb_datum *data)
 248 {
 249     size_t i;
 250
 251     for (i = 0; i < mt->n_columns; i++) {
 252         const struct ovsdb_column *c = mt->columns[i].column;
 253         const struct ovsdb_datum *src = &row->fields[c->index];
 254         struct ovsdb_datum *dst = &data[i];
 255         const struct ovsdb_type *type = &c->type;
 256
 257         if (!ovsdb_datum_equals(src, dst, type)) {
 258             ovsdb_datum_destroy(dst, type);
 259             ovsdb_datum_clone(dst, src, type);
 260         }
 261     }
 262 }
 263
 264 /* Frees all of the mt->n_columns ovsdb_datums in data[], using the types taken
 265  * from mt->columns[], plus 'data' itself. */
 266 static void
 267 free_monitor_row_data(const struct ovsdb_monitor_table *mt,
 268                       struct ovsdb_datum *data)
 269 {
 270     if (data) {
 271         size_t i;
 272
 273         for (i = 0; i < mt->n_columns; i++) {
 274             const struct ovsdb_column *c = mt->columns[i].column;
 275
 276             ovsdb_datum_destroy(&data[i], &c->type);
 277         }
 278         free(data);
 279     }
 280 }
 281
 282 /* Frees 'row', which must have been created from 'mt'. */
 283 static void
 284 ovsdb_monitor_row_destroy(const struct ovsdb_monitor_table *mt,
 285                           struct ovsdb_monitor_row *row)
 286 {
 287     if (row) {
 288         free_monitor_row_data(mt, row->old);
 289         free_monitor_row_data(mt, row->new);
 290         free(row);
 291     }
 292 }
 293
 294 void
 295 ovsdb_monitor_add_jsonrpc_monitor(struct ovsdb_monitor *dbmon,
 296                                   struct ovsdb_jsonrpc_monitor *jsonrpc_monitor)
 297 {
 298     struct jsonrpc_monitor_node *jm;
 299
 300     jm = xzalloc(sizeof *jm);
 301     jm->jsonrpc_monitor = jsonrpc_monitor;
 302     list_push_back(&dbmon->jsonrpc_monitors, &jm->node);
 303 }
 304
 305 struct ovsdb_monitor *
 306 ovsdb_monitor_create(struct ovsdb *db,
 307                      struct ovsdb_jsonrpc_monitor *jsonrpc_monitor)
 308 {
 309     struct ovsdb_monitor *dbmon;
 310
 311     dbmon = xzalloc(sizeof *dbmon);
 312
 313     ovsdb_replica_init(&dbmon->replica, &ovsdb_jsonrpc_replica_class);
 314     ovsdb_add_replica(db, &dbmon->replica);
 315     list_init(&dbmon->jsonrpc_monitors);
 316     dbmon->db = db;
 317     dbmon->n_transactions = 0;
 318     shash_init(&dbmon->tables);
 319     hmap_node_nullify(&dbmon->hmap_node);
 320     hmap_init(&dbmon->json_cache);
 321
 322     ovsdb_monitor_add_jsonrpc_monitor(dbmon, jsonrpc_monitor);
 323     return dbmon;
 324 }
 325
 326 void
 327 ovsdb_monitor_add_table(struct ovsdb_monitor *m,
 328                         const struct ovsdb_table *table)
 329 {
 330     struct ovsdb_monitor_table *mt;
 331
 332     mt = xzalloc(sizeof *mt);
 333     mt->table = table;
 334     shash_add(&m->tables, table->schema->name, mt);
 335     hmap_init(&mt->changes);
 336 }
 337
 338 void
 339 ovsdb_monitor_add_column(struct ovsdb_monitor *dbmon,
 340                          const struct ovsdb_table *table,
 341                          const struct ovsdb_column *column,
 342                          enum ovsdb_monitor_selection select,
 343                          size_t *allocated_columns)
 344 {
 345     struct ovsdb_monitor_table *mt;
 346     struct ovsdb_monitor_column *c;
 347
 348     mt = shash_find_data(&dbmon->tables, table->schema->name);
 349
 350     if (mt->n_columns >= *allocated_columns) {
 351         mt->columns = x2nrealloc(mt->columns, allocated_columns,
 352                                  sizeof *mt->columns);
 353     }
 354
 355     mt->select |= select;
 356     c = &mt->columns[mt->n_columns++];
 357     c->column = column;
 358     c->select = select;
 359 }
 360
 361 /* Check for duplicated column names. Return the first
 362  * duplicated column's name if found. Otherwise return
 363  * NULL.  */
 364 const char * OVS_WARN_UNUSED_RESULT
 365 ovsdb_monitor_table_check_duplicates(struct ovsdb_monitor *m,
 366                                      const struct ovsdb_table *table)
 367 {
 368     struct ovsdb_monitor_table *mt;
 369     int i;
 370
 371     mt = shash_find_data(&m->tables, table->schema->name);
 372
 373     if (mt) {
 374         /* Check for duplicate columns. */
 375         qsort(mt->columns, mt->n_columns, sizeof *mt->columns,
 376               compare_ovsdb_monitor_column);
 377         for (i = 1; i < mt->n_columns; i++) {
 378             if (mt->columns[i].column == mt->columns[i - 1].column) {
 379                 return mt->columns[i].column->name;
 380             }
 381         }
 382     }
 383
 384     return NULL;
 385 }
 386
 387 static struct ovsdb_monitor_changes *
 388 ovsdb_monitor_table_add_changes(struct ovsdb_monitor_table *mt,
 389                                 uint64_t next_txn)
 390 {
 391     struct ovsdb_monitor_changes *changes;
 392
 393     changes = xzalloc(sizeof *changes);
 394
 395     changes->transaction = next_txn;
 396     changes->mt = mt;
 397     changes->n_refs = 1;
 398     hmap_init(&changes->rows);
 399     hmap_insert(&mt->changes, &changes->hmap_node, hash_uint64(next_txn));
 400
 401     return changes;
 402 };
 403
 404 static struct ovsdb_monitor_changes *
 405 ovsdb_monitor_table_find_changes(struct ovsdb_monitor_table *mt,
 406                                  uint64_t transaction)
 407 {
 408     struct ovsdb_monitor_changes *changes;
 409     size_t hash = hash_uint64(transaction);
 410
 411     HMAP_FOR_EACH_WITH_HASH(changes, hmap_node, hash, &mt->changes) {
 412         if (changes->transaction == transaction) {
 413             return changes;
 414         }
 415     }
 416
 417     return NULL;
 418 }
 419
 420 /* Stop currently tracking changes to table 'mt' since 'transaction'. */
 421 static void
 422 ovsdb_monitor_table_untrack_changes(struct ovsdb_monitor_table *mt,
 423                                     uint64_t transaction)
 424 {
 425     struct ovsdb_monitor_changes *changes =
 426                 ovsdb_monitor_table_find_changes(mt, transaction);
 427     if (changes) {
 428         if (--changes->n_refs == 0) {
 429             hmap_remove(&mt->changes, &changes->hmap_node);
 430             ovsdb_monitor_changes_destroy(changes);
 431         }
 432     }
 433 }
 434
 435 /* Start tracking changes to table 'mt' begins from 'transaction' inclusive.
 436  */
 437 static void
 438 ovsdb_monitor_table_track_changes(struct ovsdb_monitor_table *mt,
 439                                   uint64_t transaction)
 440 {
 441     struct ovsdb_monitor_changes *changes;
 442
 443     changes = ovsdb_monitor_table_find_changes(mt, transaction);
 444     if (changes) {
 445         changes->n_refs++;
 446     } else {
 447         ovsdb_monitor_table_add_changes(mt, transaction);
 448     }
 449 }
 450
 451 static void
 452 ovsdb_monitor_changes_destroy(struct ovsdb_monitor_changes *changes)
 453 {
 454     struct ovsdb_monitor_row *row, *next;
 455
 456     HMAP_FOR_EACH_SAFE (row, next, hmap_node, &changes->rows) {
 457         hmap_remove(&changes->rows, &row->hmap_node);
 458         ovsdb_monitor_row_destroy(changes->mt, row);
 459     }
 460     hmap_destroy(&changes->rows);
 461     free(changes);
 462 }
 463
 464 static enum ovsdb_monitor_selection
 465 ovsdb_monitor_row_update_type(bool initial, const bool old, const bool new)
 466 {
 467     return initial ? OJMS_INITIAL
 468             : !old ? OJMS_INSERT
 469             : !new ? OJMS_DELETE
 470             : OJMS_MODIFY;
 471 }
 472 static bool
 473 ovsdb_monitor_row_skip_update(const struct ovsdb_monitor_table *mt,
 474                               const struct ovsdb_monitor_row *row,
 475                               enum ovsdb_monitor_selection type,
 476                               unsigned long int *changed)
 477 {
 478     if (!(mt->select & type)) {
 479         return true;
 480     }
 481
 482     if (type == OJMS_MODIFY) {
 483         size_t i, n_changes;
 484
 485         n_changes = 0;
 486         memset(changed, 0, bitmap_n_bytes(mt->n_columns));
 487         for (i = 0; i < mt->n_columns; i++) {
 488             const struct ovsdb_column *c = mt->columns[i].column;
 489             if (!ovsdb_datum_equals(&row->old[i], &row->new[i], &c->type)) {
 490                 bitmap_set1(changed, i);
 491                 n_changes++;
 492             }
 493         }
 494         if (!n_changes) {
 495             /* No actual changes: presumably a row changed and then
 496              * changed back later. */
 497             return true;
 498         }
 499     }
 500
 501     return false;
 502 }
 503
 504 /* Returns JSON for a <row-update> (as described in RFC 7047) for 'row' within
 505  * 'mt', or NULL if no row update should be sent.
 506  *
 507  * The caller should specify 'initial' as true if the returned JSON is going to
 508  * be used as part of the initial reply to a "monitor" request, false if it is
 509  * going to be used as part of an "update" notification.
 510  *
 511  * 'changed' must be a scratch buffer for internal use that is at least
 512  * bitmap_n_bytes(mt->n_columns) bytes long. */
 513 static struct json *
 514 ovsdb_monitor_compose_row_update(
 515     const struct ovsdb_monitor_table *mt,
 516     const struct ovsdb_monitor_row *row,
 517     bool initial, unsigned long int *changed)
 518 {
 519     enum ovsdb_monitor_selection type;
 520     struct json *old_json, *new_json;
 521     struct json *row_json;
 522     size_t i;
 523
 524     type = ovsdb_monitor_row_update_type(initial, row->old, row->new);
 525     if (ovsdb_monitor_row_skip_update(mt, row, type, changed)) {
 526         return NULL;
 527     }
 528
 529     row_json = json_object_create();
 530     old_json = new_json = NULL;
 531     if (type & (OJMS_DELETE | OJMS_MODIFY)) {
 532         old_json = json_object_create();
 533         json_object_put(row_json, "old", old_json);
 534     }
 535     if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
 536         new_json = json_object_create();
 537         json_object_put(row_json, "new", new_json);
 538     }
 539     for (i = 0; i < mt->n_columns; i++) {
 540         const struct ovsdb_monitor_column *c = &mt->columns[i];
 541
 542         if (!(type & c->select)) {
 543             /* We don't care about this type of change for this
 544              * particular column (but we will care about it for some
 545              * other column). */
 546             continue;
 547         }
 548
 549         if ((type == OJMS_MODIFY && bitmap_is_set(changed, i))
 550             || type == OJMS_DELETE) {
 551             json_object_put(old_json, c->column->name,
 552                             ovsdb_datum_to_json(&row->old[i],
 553                                                 &c->column->type));
 554         }
 555         if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
 556             json_object_put(new_json, c->column->name,
 557                             ovsdb_datum_to_json(&row->new[i],
 558                                                 &c->column->type));
 559         }
 560     }
 561
 562     return row_json;
 563 }
 564
 565 /* Returns JSON for a <row-update2> (as described in ovsdb-server(1) mapage)
 566  * for 'row' within * 'mt', or NULL if no row update should be sent.
 567  *
 568  * The caller should specify 'initial' as true if the returned JSON is
 569  * going to be used as part of the initial reply to a "monitor2" request,
 570  * false if it is going to be used as part of an "update2" notification.
 571  *
 572  * 'changed' must be a scratch buffer for internal use that is at least
 573  * bitmap_n_bytes(mt->n_columns) bytes long. */
 574 static struct json *
 575 ovsdb_monitor_compose_row_update2(
 576     const struct ovsdb_monitor_table *mt,
 577     const struct ovsdb_monitor_row *row,
 578     bool initial, unsigned long int *changed)
 579 {
 580     enum ovsdb_monitor_selection type;
 581     struct json *row_update2, *diff_json;
 582     size_t i;
 583
 584     type = ovsdb_monitor_row_update_type(initial, row->old, row->new);
 585     if (ovsdb_monitor_row_skip_update(mt, row, type, changed)) {
 586         return NULL;
 587     }
 588
 589     row_update2 = json_object_create();
 590     if (type == OJMS_DELETE) {
 591         json_object_put(row_update2, "delete", json_null_create());
 592     } else {
 593         diff_json = json_object_create();
 594         const char *op;
 595
 596         for (i = 0; i < mt->n_columns; i++) {
 597             const struct ovsdb_monitor_column *c = &mt->columns[i];
 598
 599             if (!(type & c->select)) {
 600                 /* We don't care about this type of change for this
 601                  * particular column (but we will care about it for some
 602                  * other column). */
 603                 continue;
 604             }
 605
 606             if (type == OJMS_MODIFY) {
 607                 struct ovsdb_datum diff;
 608
 609                 if (!bitmap_is_set(changed, i)) {
 610                     continue;
 611                 }
 612
 613                 ovsdb_datum_diff(&diff ,&row->old[i], &row->new[i],
 614                                         &c->column->type);
 615                 json_object_put(diff_json, c->column->name,
 616                                 ovsdb_datum_to_json(&diff, &c->column->type));
 617                 ovsdb_datum_destroy(&diff, &c->column->type);
 618             } else {
 619                 if (!ovsdb_datum_is_default(&row->new[i], &c->column->type)) {
 620                     json_object_put(diff_json, c->column->name,
 621                                     ovsdb_datum_to_json(&row->new[i],
 622                                                         &c->column->type));
 623                 }
 624             }
 625         }
 626
 627         op = type == OJMS_INITIAL ? "initial"
 628                                   : type == OJMS_MODIFY ? "modify" : "insert";
 629         json_object_put(row_update2, op, diff_json);
 630     }
 631
 632     return row_update2;
 633 }
 634
 635 static size_t
 636 ovsdb_monitor_max_columns(struct ovsdb_monitor *dbmon)
 637 {
 638     struct shash_node *node;
 639     size_t max_columns = 0;
 640
 641     SHASH_FOR_EACH (node, &dbmon->tables) {
 642         struct ovsdb_monitor_table *mt = node->data;
 643
 644         max_columns = MAX(max_columns, mt->n_columns);
 645     }
 646
 647     return max_columns;
 648 }
 649
 650 /* Constructs and returns JSON for a <table-updates> object (as described in
 651  * RFC 7047) for all the outstanding changes within 'monitor', starting from
 652  * 'transaction'.  */
 653 static struct json*
 654 ovsdb_monitor_compose_update(struct ovsdb_monitor *dbmon,
 655                              bool initial, uint64_t transaction,
 656                              compose_row_update_cb_func row_update)
 657 {
 658     struct shash_node *node;
 659     struct json *json;
 660     size_t max_columns = ovsdb_monitor_max_columns(dbmon);
 661     unsigned long int *changed = xmalloc(bitmap_n_bytes(max_columns));
 662
 663     json = NULL;
 664     SHASH_FOR_EACH (node, &dbmon->tables) {
 665         struct ovsdb_monitor_table *mt = node->data;
 666         struct ovsdb_monitor_row *row, *next;
 667         struct ovsdb_monitor_changes *changes;
 668         struct json *table_json = NULL;
 669
 670         changes = ovsdb_monitor_table_find_changes(mt, transaction);
 671         if (!changes) {
 672             continue;
 673         }
 674
 675         HMAP_FOR_EACH_SAFE (row, next, hmap_node, &changes->rows) {
 676             struct json *row_json;
 677
 678             row_json = (*row_update)(mt, row, initial, changed);
 679             if (row_json) {
 680                 char uuid[UUID_LEN + 1];
 681
 682                 /* Create JSON object for transaction overall. */
 683                 if (!json) {
 684                     json = json_object_create();
 685                 }
 686
 687                 /* Create JSON object for transaction on this table. */
 688                 if (!table_json) {
 689                     table_json = json_object_create();
 690                     json_object_put(json, mt->table->schema->name, table_json);
 691                 }
 692
 693                 /* Add JSON row to JSON table. */
 694                 snprintf(uuid, sizeof uuid, UUID_FMT, UUID_ARGS(&row->uuid));
 695                 json_object_put(table_json, uuid, row_json);
 696             }
 697         }
 698     }
 699     free(changed);
 700
 701     return json;
 702 }
 703
 704 /* Returns JSON for a <table-updates> object (as described in RFC 7047)
 705  * for all the outstanding changes within 'monitor' that starts from
 706  * '*unflushed' transaction id.
 707  *
 708  * The caller should specify 'initial' as true if the returned JSON is going to
 709  * be used as part of the initial reply to a "monitor" request, false if it is
 710  * going to be used as part of an "update" notification. */
 711 struct json *
 712 ovsdb_monitor_get_update(struct ovsdb_monitor *dbmon,
 713                          bool initial, uint64_t *unflushed,
 714                          enum ovsdb_monitor_version version)
 715 {
 716     struct ovsdb_monitor_json_cache_node *cache_node;
 717     struct shash_node *node;
 718     struct json *json;
 719     uint64_t prev_txn = *unflushed;
 720     uint64_t next_txn = dbmon->n_transactions + 1;
 721
 722     /* Return a clone of cached json if one exists. Otherwise,
 723      * generate a new one and add it to the cache.  */
 724     cache_node = ovsdb_monitor_json_cache_search(dbmon, prev_txn);
 725     if (cache_node) {
 726         json = cache_node->json ? json_clone(cache_node->json) : NULL;
 727     } else {
 728         if (version == OVSDB_MONITOR_V1) {
 729             json = ovsdb_monitor_compose_update(dbmon, initial, prev_txn,
 730                                         ovsdb_monitor_compose_row_update);
 731         } else {
 732             ovs_assert(version == OVSDB_MONITOR_V2);
 733             json = ovsdb_monitor_compose_update(dbmon, initial, prev_txn,
 734                                         ovsdb_monitor_compose_row_update2);
 735         }
 736         ovsdb_monitor_json_cache_insert(dbmon, prev_txn, json);
 737     }
 738
 739     /* Maintain transaction id of 'changes'. */
 740     SHASH_FOR_EACH (node, &dbmon->tables) {
 741         struct ovsdb_monitor_table *mt = node->data;
 742
 743         ovsdb_monitor_table_untrack_changes(mt, prev_txn);
 744         ovsdb_monitor_table_track_changes(mt, next_txn);
 745     }
 746     *unflushed = next_txn;
 747
 748     return json;
 749 }
 750
 751 bool
 752 ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon,
 753                           uint64_t next_transaction)
 754 {
 755     ovs_assert(next_transaction <= dbmon->n_transactions + 1);
 756     return (next_transaction <= dbmon->n_transactions);
 757 }
 758
 759 void
 760 ovsdb_monitor_table_add_select(struct ovsdb_monitor *dbmon,
 761                                const struct ovsdb_table *table,
 762                                enum ovsdb_monitor_selection select)
 763 {
 764     struct ovsdb_monitor_table * mt;
 765
 766     mt = shash_find_data(&dbmon->tables, table->schema->name);
 767     mt->select |= select;
 768 }
 769
 770  /*
 771  * If a row's change type (insert, delete or modify) matches that of
 772  * the monitor, they should be sent to the monitor's clients as updates.
 773  * Of cause, the monitor should also internally update with this change.
 774  *
 775  * When a change type does not require client side update, the monitor
 776  * may still need to keep track of certain changes in order to generate
 777  * correct future updates.  For example, the monitor internal state should
 778  * be updated whenever a new row is inserted, in order to generate the
 779  * correct initial state, regardless if a insert change type is being
 780  * monitored.
 781  *
 782  * On the other hand, if a transaction only contains changes to columns
 783  * that are not monitored, this transaction can be safely ignored by the
 784  * monitor.
 785  *
 786  * Thus, the order of the declaration is important:
 787  * 'OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE' always implies
 788  * 'OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE', but not vice versa.  */
 789 enum ovsdb_monitor_changes_efficacy {
 790     OVSDB_CHANGES_NO_EFFECT,                /* Monitor does not care about this
 791                                                change.  */
 792     OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE,  /* Monitor internal updates. */
 793     OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE,  /* Client needs to be updated.  */
 794 };
 795
 796 struct ovsdb_monitor_aux {
 797     const struct ovsdb_monitor *monitor;
 798     struct ovsdb_monitor_table *mt;
 799     enum ovsdb_monitor_changes_efficacy efficacy;
 800 };
 801
 802 static void
 803 ovsdb_monitor_init_aux(struct ovsdb_monitor_aux *aux,
 804                        const struct ovsdb_monitor *m)
 805 {
 806     aux->monitor = m;
 807     aux->mt = NULL;
 808     aux->efficacy = OVSDB_CHANGES_NO_EFFECT;
 809 }
 810
 811 static void
 812 ovsdb_monitor_changes_update(const struct ovsdb_row *old,
 813                              const struct ovsdb_row *new,
 814                              const struct ovsdb_monitor_table *mt,
 815                              struct ovsdb_monitor_changes *changes)
 816 {
 817     const struct uuid *uuid = ovsdb_row_get_uuid(new ? new : old);
 818     struct ovsdb_monitor_row *change;
 819
 820     change = ovsdb_monitor_changes_row_find(changes, uuid);
 821     if (!change) {
 822         change = xzalloc(sizeof *change);
 823         hmap_insert(&changes->rows, &change->hmap_node, uuid_hash(uuid));
 824         change->uuid = *uuid;
 825         change->old = clone_monitor_row_data(mt, old);
 826         change->new = clone_monitor_row_data(mt, new);
 827     } else {
 828         if (new) {
 829             update_monitor_row_data(mt, new, change->new);
 830         } else {
 831             free_monitor_row_data(mt, change->new);
 832             change->new = NULL;
 833
 834             if (!change->old) {
 835                 /* This row was added then deleted.  Forget about it. */
 836                 hmap_remove(&changes->rows, &change->hmap_node);
 837                 free(change);
 838             }
 839         }
 840     }
 841 }
 842
 843 static bool
 844 ovsdb_monitor_columns_changed(const struct ovsdb_monitor_table *mt,
 845                               const unsigned long int *changed)
 846 {
 847     size_t i;
 848
 849     for (i = 0; i < mt->n_columns; i++) {
 850         size_t column_index = mt->columns[i].column->index;
 851
 852         if (bitmap_is_set(changed, column_index)) {
 853             return true;
 854         }
 855     }
 856
 857     return false;
 858 }
 859
 860 /* Return the efficacy of a row's change to a monitor table.
 861  *
 862  * Please see the block comment above 'ovsdb_monitor_changes_efficacy'
 863  * definition form more information.  */
 864 static enum ovsdb_monitor_changes_efficacy
 865 ovsdb_monitor_changes_classify(enum ovsdb_monitor_selection type,
 866                                const struct ovsdb_monitor_table *mt,
 867                                const unsigned long int *changed)
 868 {
 869     if (type == OJMS_MODIFY &&
 870         !ovsdb_monitor_columns_changed(mt, changed)) {
 871         return OVSDB_CHANGES_NO_EFFECT;
 872     }
 873
 874     return (mt->select & type)
 875                 ?  OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE
 876                 :  OVSDB_CHANGES_REQUIRE_INTERNAL_UPDATE;
 877 }
 878
 879 static bool
 880 ovsdb_monitor_change_cb(const struct ovsdb_row *old,
 881                         const struct ovsdb_row *new,
 882                         const unsigned long int *changed,
 883                         void *aux_)
 884 {
 885     struct ovsdb_monitor_aux *aux = aux_;
 886     const struct ovsdb_monitor *m = aux->monitor;
 887     struct ovsdb_table *table = new ? new->table : old->table;
 888     struct ovsdb_monitor_table *mt;
 889     struct ovsdb_monitor_changes *changes;
 890
 891     if (!aux->mt || table != aux->mt->table) {
 892         aux->mt = shash_find_data(&m->tables, table->schema->name);
 893         if (!aux->mt) {
 894             /* We don't care about rows in this table at all.  Tell the caller
 895              * to skip it.  */
 896             return false;
 897         }
 898     }
 899     mt = aux->mt;
 900
 901     HMAP_FOR_EACH(changes, hmap_node, &mt->changes) {
 902         enum ovsdb_monitor_changes_efficacy efficacy;
 903         enum ovsdb_monitor_selection type;
 904
 905         type = ovsdb_monitor_row_update_type(false, old, new);
 906         efficacy = ovsdb_monitor_changes_classify(type, mt, changed);
 907         if (efficacy > OVSDB_CHANGES_NO_EFFECT) {
 908             ovsdb_monitor_changes_update(old, new, mt, changes);
 909         }
 910
 911         if (aux->efficacy < efficacy) {
 912             aux->efficacy = efficacy;
 913         }
 914     }
 915
 916     return true;
 917 }
 918
 919 void
 920 ovsdb_monitor_get_initial(const struct ovsdb_monitor *dbmon)
 921 {
 922     struct ovsdb_monitor_aux aux;
 923     struct shash_node *node;
 924
 925     ovsdb_monitor_init_aux(&aux, dbmon);
 926     SHASH_FOR_EACH (node, &dbmon->tables) {
 927         struct ovsdb_monitor_table *mt = node->data;
 928
 929         if (mt->select & OJMS_INITIAL) {
 930             struct ovsdb_row *row;
 931             struct ovsdb_monitor_changes *changes;
 932
 933             changes = ovsdb_monitor_table_find_changes(mt, 0);
 934             if (!changes) {
 935                 changes = ovsdb_monitor_table_add_changes(mt, 0);
 936                 HMAP_FOR_EACH (row, hmap_node, &mt->table->rows) {
 937                     ovsdb_monitor_changes_update(NULL, row, mt, changes);
 938                 }
 939             } else {
 940                 changes->n_refs++;
 941             }
 942         }
 943     }
 944 }
 945
 946 void
 947 ovsdb_monitor_remove_jsonrpc_monitor(struct ovsdb_monitor *dbmon,
 948                    struct ovsdb_jsonrpc_monitor *jsonrpc_monitor)
 949 {
 950     struct jsonrpc_monitor_node *jm;
 951
 952     if (list_is_empty(&dbmon->jsonrpc_monitors)) {
 953         ovsdb_monitor_destroy(dbmon);
 954         return;
 955     }
 956
 957     /* Find and remove the jsonrpc monitor from the list.  */
 958     LIST_FOR_EACH(jm, node, &dbmon->jsonrpc_monitors) {
 959         if (jm->jsonrpc_monitor == jsonrpc_monitor) {
 960             list_remove(&jm->node);
 961             free(jm);
 962
 963             /* Destroy ovsdb monitor if this is the last user.  */
 964             if (list_is_empty(&dbmon->jsonrpc_monitors)) {
 965                 ovsdb_monitor_destroy(dbmon);
 966             }
 967
 968             return;
 969         };
 970     }
 971
 972     /* Should never reach here. jsonrpc_monitor should be on the list.  */
 973     OVS_NOT_REACHED();
 974 }
 975
 976 static bool
 977 ovsdb_monitor_table_equal(const struct ovsdb_monitor_table *a,
 978                           const struct ovsdb_monitor_table *b)
 979 {
 980     size_t i;
 981
 982     if ((a->table != b->table) ||
 983         (a->select != b->select) ||
 984         (a->n_columns != b->n_columns)) {
 985         return false;
 986     }
 987
 988     for (i = 0; i < a->n_columns; i++) {
 989         if ((a->columns[i].column != b->columns[i].column) ||
 990             (a->columns[i].select != b->columns[i].select)) {
 991             return false;
 992         }
 993     }
 994
 995     return true;
 996 }
 997
 998 static bool
 999 ovsdb_monitor_equal(const struct ovsdb_monitor *a,
1000                     const struct ovsdb_monitor *b)
1001 {
1002     struct shash_node *node;
1003
1004     if (shash_count(&a->tables) != shash_count(&b->tables)) {
1005         return false;
1006     }
1007
1008     SHASH_FOR_EACH(node, &a->tables) {
1009         const struct ovsdb_monitor_table *mta = node->data;
1010         const struct ovsdb_monitor_table *mtb;
1011
1012         mtb = shash_find_data(&b->tables, node->name);
1013         if (!mtb) {
1014             return false;
1015         }
1016
1017         if (!ovsdb_monitor_table_equal(mta, mtb)) {
1018             return false;
1019         }
1020     }
1021
1022     return true;
1023 }
1024
1025 static size_t
1026 ovsdb_monitor_hash(const struct ovsdb_monitor *dbmon, size_t basis)
1027 {
1028     const struct shash_node **nodes;
1029     size_t i, j, n;
1030
1031     nodes = shash_sort(&dbmon->tables);
1032     n = shash_count(&dbmon->tables);
1033
1034     for (i = 0; i < n; i++) {
1035         struct ovsdb_monitor_table *mt = nodes[i]->data;
1036
1037         basis = hash_pointer(mt->table, basis);
1038         basis = hash_3words(mt->select, mt->n_columns, basis);
1039
1040         for (j = 0; j < mt->n_columns; j++) {
1041             basis = hash_pointer(mt->columns[j].column, basis);
1042             basis = hash_2words(mt->columns[j].select, basis);
1043         }
1044     }
1045     free(nodes);
1046
1047     return basis;
1048 }
1049
1050 struct ovsdb_monitor *
1051 ovsdb_monitor_add(struct ovsdb_monitor *new_dbmon)
1052 {
1053     struct ovsdb_monitor *dbmon;
1054     size_t hash;
1055
1056     /* New_dbmon should be associated with only one jsonrpc
1057      * connections.  */
1058     ovs_assert(list_is_singleton(&new_dbmon->jsonrpc_monitors));
1059
1060     hash = ovsdb_monitor_hash(new_dbmon, 0);
1061     HMAP_FOR_EACH_WITH_HASH(dbmon, hmap_node, hash, &ovsdb_monitors) {
1062         if (ovsdb_monitor_equal(dbmon,  new_dbmon)) {
1063             return dbmon;
1064         }
1065     }
1066
1067     hmap_insert(&ovsdb_monitors, &new_dbmon->hmap_node, hash);
1068     return new_dbmon;
1069 }
1070
1071 static void
1072 ovsdb_monitor_destroy(struct ovsdb_monitor *dbmon)
1073 {
1074     struct shash_node *node;
1075
1076     list_remove(&dbmon->replica.node);
1077
1078     if (!hmap_node_is_null(&dbmon->hmap_node)) {
1079         hmap_remove(&ovsdb_monitors, &dbmon->hmap_node);
1080     }
1081
1082     ovsdb_monitor_json_cache_flush(dbmon);
1083     hmap_destroy(&dbmon->json_cache);
1084
1085     SHASH_FOR_EACH (node, &dbmon->tables) {
1086         struct ovsdb_monitor_table *mt = node->data;
1087         struct ovsdb_monitor_changes *changes, *next;
1088
1089         HMAP_FOR_EACH_SAFE (changes, next, hmap_node, &mt->changes) {
1090             hmap_remove(&mt->changes, &changes->hmap_node);
1091             ovsdb_monitor_changes_destroy(changes);
1092         }
1093         hmap_destroy(&mt->changes);
1094         free(mt->columns);
1095         free(mt);
1096     }
1097     shash_destroy(&dbmon->tables);
1098     free(dbmon);
1099 }
1100
1101 static struct ovsdb_error *
1102 ovsdb_monitor_commit(struct ovsdb_replica *replica,
1103                      const struct ovsdb_txn *txn,
1104                      bool durable OVS_UNUSED)
1105 {
1106     struct ovsdb_monitor *m = ovsdb_monitor_cast(replica);
1107     struct ovsdb_monitor_aux aux;
1108
1109     ovsdb_monitor_init_aux(&aux, m);
1110     ovsdb_txn_for_each_change(txn, ovsdb_monitor_change_cb, &aux);
1111
1112     if (aux.efficacy == OVSDB_CHANGES_REQUIRE_EXTERNAL_UPDATE) {
1113         ovsdb_monitor_json_cache_flush(m);
1114         m->n_transactions++;
1115     }
1116
1117     return NULL;
1118 }
1119
1120 static void
1121 ovsdb_monitor_destroy_callback(struct ovsdb_replica *replica)
1122 {
1123     struct ovsdb_monitor *dbmon = ovsdb_monitor_cast(replica);
1124     struct jsonrpc_monitor_node *jm, *next;
1125
1126     /* Delete all front end monitors. Removing the last front
1127      * end monitor will also destroy the corresponding 'ovsdb_monitor'.
1128      * ovsdb monitor will also be destroied.  */
1129     LIST_FOR_EACH_SAFE(jm, next, node, &dbmon->jsonrpc_monitors) {
1130         ovsdb_jsonrpc_monitor_destroy(jm->jsonrpc_monitor);
1131     }
1132 }
1133
1134 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class = {
1135     ovsdb_monitor_commit,
1136     ovsdb_monitor_destroy_callback,
1137 };