ovsdb/jsonrpc-server.c

   1 /* Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014, 2015 Nicira, Inc.
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <config.h>
  17
  18 #include "jsonrpc-server.h"
  19
  20 #include <errno.h>
  21
  22 #include "bitmap.h"
  23 #include "column.h"
  24 #include "dynamic-string.h"
  25 #include "json.h"
  26 #include "jsonrpc.h"
  27 #include "ovsdb-error.h"
  28 #include "ovsdb-parser.h"
  29 #include "ovsdb.h"
  30 #include "poll-loop.h"
  31 #include "reconnect.h"
  32 #include "row.h"
  33 #include "server.h"
  34 #include "simap.h"
  35 #include "stream.h"
  36 #include "table.h"
  37 #include "timeval.h"
  38 #include "transaction.h"
  39 #include "trigger.h"
  40 #include "openvswitch/vlog.h"
  41
  42 VLOG_DEFINE_THIS_MODULE(ovsdb_jsonrpc_server);
  43
  44 struct ovsdb_jsonrpc_remote;
  45 struct ovsdb_jsonrpc_session;
  46 struct ovsdb_jsonrpc_monitor;
  47
  48 /* Message rate-limiting. */
  49 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
  50
  51 /* Sessions. */
  52 static struct ovsdb_jsonrpc_session *ovsdb_jsonrpc_session_create(
  53     struct ovsdb_jsonrpc_remote *, struct jsonrpc_session *);
  54 static void ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_remote *);
  55 static void ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_remote *);
  56 static void ovsdb_jsonrpc_session_get_memory_usage_all(
  57     const struct ovsdb_jsonrpc_remote *, struct simap *usage);
  58 static void ovsdb_jsonrpc_session_close_all(struct ovsdb_jsonrpc_remote *);
  59 static void ovsdb_jsonrpc_session_reconnect_all(struct ovsdb_jsonrpc_remote *);
  60 static void ovsdb_jsonrpc_session_set_all_options(
  61     struct ovsdb_jsonrpc_remote *, const struct ovsdb_jsonrpc_options *);
  62 static bool ovsdb_jsonrpc_session_get_status(
  63     const struct ovsdb_jsonrpc_remote *,
  64     struct ovsdb_jsonrpc_remote_status *);
  65 static void ovsdb_jsonrpc_session_unlock_all(struct ovsdb_jsonrpc_session *);
  66 static void ovsdb_jsonrpc_session_unlock__(struct ovsdb_lock_waiter *);
  67 static void ovsdb_jsonrpc_session_send(struct ovsdb_jsonrpc_session *,
  68                                        struct jsonrpc_msg *);
  69
  70 /* Triggers. */
  71 static void ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *,
  72                                          struct ovsdb *,
  73                                          struct json *id, struct json *params);
  74 static struct ovsdb_jsonrpc_trigger *ovsdb_jsonrpc_trigger_find(
  75     struct ovsdb_jsonrpc_session *, const struct json *id, size_t hash);
  76 static void ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *);
  77 static void ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *);
  78 static void ovsdb_jsonrpc_trigger_complete_done(
  79     struct ovsdb_jsonrpc_session *);
  80
  81 /* Monitors. */
  82 static struct jsonrpc_msg *ovsdb_jsonrpc_monitor_create(
  83     struct ovsdb_jsonrpc_session *, struct ovsdb *, struct json *params,
  84     const struct json *request_id);
  85 static void ovsdb_jsonrpc_monitor_destroy(struct ovsdb_jsonrpc_monitor *);
  86 static struct jsonrpc_msg *ovsdb_jsonrpc_monitor_cancel(
  87     struct ovsdb_jsonrpc_session *,
  88     struct json_array *params,
  89     const struct json *request_id);
  90 static void ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *);
  91 static void ovsdb_jsonrpc_monitor_flush_all(struct ovsdb_jsonrpc_session *);
  92 static bool ovsdb_jsonrpc_monitor_needs_flush(struct ovsdb_jsonrpc_session *);
  93 \f
  94 /* JSON-RPC database server. */
  95
  96 struct ovsdb_jsonrpc_server {
  97     struct ovsdb_server up;
  98     unsigned int n_sessions, max_sessions;
  99     struct shash remotes;      /* Contains "struct ovsdb_jsonrpc_remote *"s. */
 100 };
 101
 102 /* A configured remote.  This is either a passive stream listener plus a list
 103  * of the currently connected sessions, or a list of exactly one active
 104  * session. */
 105 struct ovsdb_jsonrpc_remote {
 106     struct ovsdb_jsonrpc_server *server;
 107     struct pstream *listener;   /* Listener, if passive. */
 108     struct ovs_list sessions;   /* List of "struct ovsdb_jsonrpc_session"s. */
 109     uint8_t dscp;
 110 };
 111
 112 static struct ovsdb_jsonrpc_remote *ovsdb_jsonrpc_server_add_remote(
 113     struct ovsdb_jsonrpc_server *, const char *name,
 114     const struct ovsdb_jsonrpc_options *options
 115 );
 116 static void ovsdb_jsonrpc_server_del_remote(struct shash_node *);
 117
 118 /* Creates and returns a new server to provide JSON-RPC access to an OVSDB.
 119  *
 120  * The caller must call ovsdb_jsonrpc_server_add_db() for each database to
 121  * which 'server' should provide access. */
 122 struct ovsdb_jsonrpc_server *
 123 ovsdb_jsonrpc_server_create(void)
 124 {
 125     struct ovsdb_jsonrpc_server *server = xzalloc(sizeof *server);
 126     ovsdb_server_init(&server->up);
 127     server->max_sessions = 330;   /* Random limit. */
 128     shash_init(&server->remotes);
 129     return server;
 130 }
 131
 132 /* Adds 'db' to the set of databases served out by 'svr'.  Returns true if
 133  * successful, false if 'db''s name is the same as some database already in
 134  * 'server'. */
 135 bool
 136 ovsdb_jsonrpc_server_add_db(struct ovsdb_jsonrpc_server *svr, struct ovsdb *db)
 137 {
 138     /* The OVSDB protocol doesn't have a way to notify a client that a
 139      * database has been added.  If some client tried to use the database
 140      * that we're adding and failed, then forcing it to reconnect seems like
 141      * a reasonable way to make it try again.
 142      *
 143      * If this is too big of a hammer in practice, we could be more selective,
 144      * e.g. disconnect only connections that actually tried to use a database
 145      * with 'db''s name. */
 146     ovsdb_jsonrpc_server_reconnect(svr);
 147
 148     return ovsdb_server_add_db(&svr->up, db);
 149 }
 150
 151 /* Removes 'db' from the set of databases served out by 'svr'.  Returns
 152  * true if successful, false if there is no database associated with 'db'. */
 153 bool
 154 ovsdb_jsonrpc_server_remove_db(struct ovsdb_jsonrpc_server *svr,
 155                                struct ovsdb *db)
 156 {
 157     /* There might be pointers to 'db' from 'svr', such as monitors or
 158      * outstanding transactions.  Disconnect all JSON-RPC connections to avoid
 159      * accesses to freed memory.
 160      *
 161      * If this is too big of a hammer in practice, we could be more selective,
 162      * e.g. disconnect only connections that actually reference 'db'. */
 163     ovsdb_jsonrpc_server_reconnect(svr);
 164
 165     return ovsdb_server_remove_db(&svr->up, db);
 166 }
 167
 168 void
 169 ovsdb_jsonrpc_server_destroy(struct ovsdb_jsonrpc_server *svr)
 170 {
 171     struct shash_node *node, *next;
 172
 173     SHASH_FOR_EACH_SAFE (node, next, &svr->remotes) {
 174         ovsdb_jsonrpc_server_del_remote(node);
 175     }
 176     shash_destroy(&svr->remotes);
 177     ovsdb_server_destroy(&svr->up);
 178     free(svr);
 179 }
 180
 181 struct ovsdb_jsonrpc_options *
 182 ovsdb_jsonrpc_default_options(const char *target)
 183 {
 184     struct ovsdb_jsonrpc_options *options = xzalloc(sizeof *options);
 185     options->max_backoff = RECONNECT_DEFAULT_MAX_BACKOFF;
 186     options->probe_interval = (stream_or_pstream_needs_probes(target)
 187                                ? RECONNECT_DEFAULT_PROBE_INTERVAL
 188                                : 0);
 189     return options;
 190 }
 191
 192 /* Sets 'svr''s current set of remotes to the names in 'new_remotes', with
 193  * options in the struct ovsdb_jsonrpc_options supplied as the data values.
 194  *
 195  * A remote is an active or passive stream connection method, e.g. "pssl:" or
 196  * "tcp:1.2.3.4". */
 197 void
 198 ovsdb_jsonrpc_server_set_remotes(struct ovsdb_jsonrpc_server *svr,
 199                                  const struct shash *new_remotes)
 200 {
 201     struct shash_node *node, *next;
 202
 203     SHASH_FOR_EACH_SAFE (node, next, &svr->remotes) {
 204         struct ovsdb_jsonrpc_remote *remote = node->data;
 205         struct ovsdb_jsonrpc_options *options
 206             = shash_find_data(new_remotes, node->name);
 207
 208         if (!options) {
 209             VLOG_INFO("%s: remote deconfigured", node->name);
 210             ovsdb_jsonrpc_server_del_remote(node);
 211         } else if (options->dscp != remote->dscp) {
 212             ovsdb_jsonrpc_server_del_remote(node);
 213          }
 214     }
 215     SHASH_FOR_EACH (node, new_remotes) {
 216         const struct ovsdb_jsonrpc_options *options = node->data;
 217         struct ovsdb_jsonrpc_remote *remote;
 218
 219         remote = shash_find_data(&svr->remotes, node->name);
 220         if (!remote) {
 221             remote = ovsdb_jsonrpc_server_add_remote(svr, node->name, options);
 222             if (!remote) {
 223                 continue;
 224             }
 225         }
 226
 227         ovsdb_jsonrpc_session_set_all_options(remote, options);
 228     }
 229 }
 230
 231 static struct ovsdb_jsonrpc_remote *
 232 ovsdb_jsonrpc_server_add_remote(struct ovsdb_jsonrpc_server *svr,
 233                                 const char *name,
 234                                 const struct ovsdb_jsonrpc_options *options)
 235 {
 236     struct ovsdb_jsonrpc_remote *remote;
 237     struct pstream *listener;
 238     int error;
 239
 240     error = jsonrpc_pstream_open(name, &listener, options->dscp);
 241     if (error && error != EAFNOSUPPORT) {
 242         VLOG_ERR_RL(&rl, "%s: listen failed: %s", name, ovs_strerror(error));
 243         return NULL;
 244     }
 245
 246     remote = xmalloc(sizeof *remote);
 247     remote->server = svr;
 248     remote->listener = listener;
 249     list_init(&remote->sessions);
 250     remote->dscp = options->dscp;
 251     shash_add(&svr->remotes, name, remote);
 252
 253     if (!listener) {
 254         ovsdb_jsonrpc_session_create(remote, jsonrpc_session_open(name, true));
 255     }
 256     return remote;
 257 }
 258
 259 static void
 260 ovsdb_jsonrpc_server_del_remote(struct shash_node *node)
 261 {
 262     struct ovsdb_jsonrpc_remote *remote = node->data;
 263
 264     ovsdb_jsonrpc_session_close_all(remote);
 265     pstream_close(remote->listener);
 266     shash_delete(&remote->server->remotes, node);
 267     free(remote);
 268 }
 269
 270 /* Stores status information for the remote named 'target', which should have
 271  * been configured on 'svr' with a call to ovsdb_jsonrpc_server_set_remotes(),
 272  * into '*status'.  On success returns true, on failure (if 'svr' doesn't have
 273  * a remote named 'target' or if that remote is an inbound remote that has no
 274  * active connections) returns false.  On failure, 'status' will be zeroed.
 275  */
 276 bool
 277 ovsdb_jsonrpc_server_get_remote_status(
 278     const struct ovsdb_jsonrpc_server *svr, const char *target,
 279     struct ovsdb_jsonrpc_remote_status *status)
 280 {
 281     const struct ovsdb_jsonrpc_remote *remote;
 282
 283     memset(status, 0, sizeof *status);
 284
 285     remote = shash_find_data(&svr->remotes, target);
 286     return remote && ovsdb_jsonrpc_session_get_status(remote, status);
 287 }
 288
 289 void
 290 ovsdb_jsonrpc_server_free_remote_status(
 291     struct ovsdb_jsonrpc_remote_status *status)
 292 {
 293     free(status->locks_held);
 294     free(status->locks_waiting);
 295     free(status->locks_lost);
 296 }
 297
 298 /* Forces all of the JSON-RPC sessions managed by 'svr' to disconnect and
 299  * reconnect. */
 300 void
 301 ovsdb_jsonrpc_server_reconnect(struct ovsdb_jsonrpc_server *svr)
 302 {
 303     struct shash_node *node;
 304
 305     SHASH_FOR_EACH (node, &svr->remotes) {
 306         struct ovsdb_jsonrpc_remote *remote = node->data;
 307
 308         ovsdb_jsonrpc_session_reconnect_all(remote);
 309     }
 310 }
 311
 312 void
 313 ovsdb_jsonrpc_server_run(struct ovsdb_jsonrpc_server *svr)
 314 {
 315     struct shash_node *node;
 316
 317     SHASH_FOR_EACH (node, &svr->remotes) {
 318         struct ovsdb_jsonrpc_remote *remote = node->data;
 319
 320         if (remote->listener) {
 321             if (svr->n_sessions < svr->max_sessions) {
 322                 struct stream *stream;
 323                 int error;
 324
 325                 error = pstream_accept(remote->listener, &stream);
 326                 if (!error) {
 327                     struct jsonrpc_session *js;
 328                     js = jsonrpc_session_open_unreliably(jsonrpc_open(stream),
 329                                                          remote->dscp);
 330                     ovsdb_jsonrpc_session_create(remote, js);
 331                 } else if (error != EAGAIN) {
 332                     VLOG_WARN_RL(&rl, "%s: accept failed: %s",
 333                                  pstream_get_name(remote->listener),
 334                                  ovs_strerror(error));
 335                 }
 336             } else {
 337                 VLOG_WARN_RL(&rl, "%s: connection exceeded maximum (%d)",
 338                              pstream_get_name(remote->listener),
 339                              svr->max_sessions);
 340             }
 341         }
 342
 343         ovsdb_jsonrpc_session_run_all(remote);
 344     }
 345 }
 346
 347 void
 348 ovsdb_jsonrpc_server_wait(struct ovsdb_jsonrpc_server *svr)
 349 {
 350     struct shash_node *node;
 351
 352     SHASH_FOR_EACH (node, &svr->remotes) {
 353         struct ovsdb_jsonrpc_remote *remote = node->data;
 354
 355         if (remote->listener && svr->n_sessions < svr->max_sessions) {
 356             pstream_wait(remote->listener);
 357         }
 358
 359         ovsdb_jsonrpc_session_wait_all(remote);
 360     }
 361 }
 362
 363 /* Adds some memory usage statistics for 'svr' into 'usage', for use with
 364  * memory_report(). */
 365 void
 366 ovsdb_jsonrpc_server_get_memory_usage(const struct ovsdb_jsonrpc_server *svr,
 367                                       struct simap *usage)
 368 {
 369     struct shash_node *node;
 370
 371     simap_increase(usage, "sessions", svr->n_sessions);
 372     SHASH_FOR_EACH (node, &svr->remotes) {
 373         struct ovsdb_jsonrpc_remote *remote = node->data;
 374
 375         ovsdb_jsonrpc_session_get_memory_usage_all(remote, usage);
 376     }
 377 }
 378 \f
 379 /* JSON-RPC database server session. */
 380
 381 struct ovsdb_jsonrpc_session {
 382     struct ovs_list node;       /* Element in remote's sessions list. */
 383     struct ovsdb_session up;
 384     struct ovsdb_jsonrpc_remote *remote;
 385
 386     /* Triggers. */
 387     struct hmap triggers;       /* Hmap of "struct ovsdb_jsonrpc_trigger"s. */
 388
 389     /* Monitors. */
 390     struct hmap monitors;       /* Hmap of "struct ovsdb_jsonrpc_monitor"s. */
 391
 392     /* Network connectivity. */
 393     struct jsonrpc_session *js;  /* JSON-RPC session. */
 394     unsigned int js_seqno;       /* Last jsonrpc_session_get_seqno() value. */
 395 };
 396
 397 static void ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *);
 398 static int ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *);
 399 static void ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *);
 400 static void ovsdb_jsonrpc_session_get_memory_usage(
 401     const struct ovsdb_jsonrpc_session *, struct simap *usage);
 402 static void ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *,
 403                                              struct jsonrpc_msg *);
 404 static void ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *,
 405                                              struct jsonrpc_msg *);
 406
 407 static struct ovsdb_jsonrpc_session *
 408 ovsdb_jsonrpc_session_create(struct ovsdb_jsonrpc_remote *remote,
 409                              struct jsonrpc_session *js)
 410 {
 411     struct ovsdb_jsonrpc_session *s;
 412
 413     s = xzalloc(sizeof *s);
 414     ovsdb_session_init(&s->up, &remote->server->up);
 415     s->remote = remote;
 416     list_push_back(&remote->sessions, &s->node);
 417     hmap_init(&s->triggers);
 418     hmap_init(&s->monitors);
 419     s->js = js;
 420     s->js_seqno = jsonrpc_session_get_seqno(js);
 421
 422     remote->server->n_sessions++;
 423
 424     return s;
 425 }
 426
 427 static void
 428 ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *s)
 429 {
 430     ovsdb_jsonrpc_monitor_remove_all(s);
 431     ovsdb_jsonrpc_session_unlock_all(s);
 432     ovsdb_jsonrpc_trigger_complete_all(s);
 433
 434     hmap_destroy(&s->monitors);
 435     hmap_destroy(&s->triggers);
 436
 437     jsonrpc_session_close(s->js);
 438     list_remove(&s->node);
 439     s->remote->server->n_sessions--;
 440     ovsdb_session_destroy(&s->up);
 441     free(s);
 442 }
 443
 444 static int
 445 ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *s)
 446 {
 447     jsonrpc_session_run(s->js);
 448     if (s->js_seqno != jsonrpc_session_get_seqno(s->js)) {
 449         s->js_seqno = jsonrpc_session_get_seqno(s->js);
 450         ovsdb_jsonrpc_trigger_complete_all(s);
 451         ovsdb_jsonrpc_monitor_remove_all(s);
 452         ovsdb_jsonrpc_session_unlock_all(s);
 453     }
 454
 455     ovsdb_jsonrpc_trigger_complete_done(s);
 456
 457     if (!jsonrpc_session_get_backlog(s->js)) {
 458         struct jsonrpc_msg *msg;
 459
 460         ovsdb_jsonrpc_monitor_flush_all(s);
 461
 462         msg = jsonrpc_session_recv(s->js);
 463         if (msg) {
 464             if (msg->type == JSONRPC_REQUEST) {
 465                 ovsdb_jsonrpc_session_got_request(s, msg);
 466             } else if (msg->type == JSONRPC_NOTIFY) {
 467                 ovsdb_jsonrpc_session_got_notify(s, msg);
 468             } else {
 469                 VLOG_WARN("%s: received unexpected %s message",
 470                           jsonrpc_session_get_name(s->js),
 471                           jsonrpc_msg_type_to_string(msg->type));
 472                 jsonrpc_session_force_reconnect(s->js);
 473                 jsonrpc_msg_destroy(msg);
 474             }
 475         }
 476     }
 477     return jsonrpc_session_is_alive(s->js) ? 0 : ETIMEDOUT;
 478 }
 479
 480 static void
 481 ovsdb_jsonrpc_session_set_options(struct ovsdb_jsonrpc_session *session,
 482                                   const struct ovsdb_jsonrpc_options *options)
 483 {
 484     jsonrpc_session_set_max_backoff(session->js, options->max_backoff);
 485     jsonrpc_session_set_probe_interval(session->js, options->probe_interval);
 486     jsonrpc_session_set_dscp(session->js, options->dscp);
 487 }
 488
 489 static void
 490 ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_remote *remote)
 491 {
 492     struct ovsdb_jsonrpc_session *s, *next;
 493
 494     LIST_FOR_EACH_SAFE (s, next, node, &remote->sessions) {
 495         int error = ovsdb_jsonrpc_session_run(s);
 496         if (error) {
 497             ovsdb_jsonrpc_session_close(s);
 498         }
 499     }
 500 }
 501
 502 static void
 503 ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *s)
 504 {
 505     jsonrpc_session_wait(s->js);
 506     if (!jsonrpc_session_get_backlog(s->js)) {
 507         if (ovsdb_jsonrpc_monitor_needs_flush(s)) {
 508             poll_immediate_wake();
 509         } else {
 510             jsonrpc_session_recv_wait(s->js);
 511         }
 512     }
 513 }
 514
 515 static void
 516 ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_remote *remote)
 517 {
 518     struct ovsdb_jsonrpc_session *s;
 519
 520     LIST_FOR_EACH (s, node, &remote->sessions) {
 521         ovsdb_jsonrpc_session_wait(s);
 522     }
 523 }
 524
 525 static void
 526 ovsdb_jsonrpc_session_get_memory_usage(const struct ovsdb_jsonrpc_session *s,
 527                                        struct simap *usage)
 528 {
 529     simap_increase(usage, "triggers", hmap_count(&s->triggers));
 530     simap_increase(usage, "monitors", hmap_count(&s->monitors));
 531     simap_increase(usage, "backlog", jsonrpc_session_get_backlog(s->js));
 532 }
 533
 534 static void
 535 ovsdb_jsonrpc_session_get_memory_usage_all(
 536     const struct ovsdb_jsonrpc_remote *remote,
 537     struct simap *usage)
 538 {
 539     struct ovsdb_jsonrpc_session *s;
 540
 541     LIST_FOR_EACH (s, node, &remote->sessions) {
 542         ovsdb_jsonrpc_session_get_memory_usage(s, usage);
 543     }
 544 }
 545
 546 static void
 547 ovsdb_jsonrpc_session_close_all(struct ovsdb_jsonrpc_remote *remote)
 548 {
 549     struct ovsdb_jsonrpc_session *s, *next;
 550
 551     LIST_FOR_EACH_SAFE (s, next, node, &remote->sessions) {
 552         ovsdb_jsonrpc_session_close(s);
 553     }
 554 }
 555
 556 /* Forces all of the JSON-RPC sessions managed by 'remote' to disconnect and
 557  * reconnect. */
 558 static void
 559 ovsdb_jsonrpc_session_reconnect_all(struct ovsdb_jsonrpc_remote *remote)
 560 {
 561     struct ovsdb_jsonrpc_session *s, *next;
 562
 563     LIST_FOR_EACH_SAFE (s, next, node, &remote->sessions) {
 564         jsonrpc_session_force_reconnect(s->js);
 565         if (!jsonrpc_session_is_alive(s->js)) {
 566             ovsdb_jsonrpc_session_close(s);
 567         }
 568     }
 569 }
 570
 571 /* Sets the options for all of the JSON-RPC sessions managed by 'remote' to
 572  * 'options'.
 573  *
 574  * (The dscp value can't be changed directly; the caller must instead close and
 575  * re-open the session.) */
 576 static void
 577 ovsdb_jsonrpc_session_set_all_options(
 578     struct ovsdb_jsonrpc_remote *remote,
 579     const struct ovsdb_jsonrpc_options *options)
 580 {
 581     struct ovsdb_jsonrpc_session *s;
 582
 583     LIST_FOR_EACH (s, node, &remote->sessions) {
 584         ovsdb_jsonrpc_session_set_options(s, options);
 585     }
 586 }
 587
 588 static bool
 589 ovsdb_jsonrpc_session_get_status(const struct ovsdb_jsonrpc_remote *remote,
 590                                  struct ovsdb_jsonrpc_remote_status *status)
 591 {
 592     const struct ovsdb_jsonrpc_session *s;
 593     const struct jsonrpc_session *js;
 594     struct ovsdb_lock_waiter *waiter;
 595     struct reconnect_stats rstats;
 596     struct ds locks_held, locks_waiting, locks_lost;
 597
 598     status->bound_port = (remote->listener
 599                           ? pstream_get_bound_port(remote->listener)
 600                           : htons(0));
 601
 602     if (list_is_empty(&remote->sessions)) {
 603         return false;
 604     }
 605     s = CONTAINER_OF(remote->sessions.next, struct ovsdb_jsonrpc_session, node);
 606     js = s->js;
 607
 608     status->is_connected = jsonrpc_session_is_connected(js);
 609     status->last_error = jsonrpc_session_get_status(js);
 610
 611     jsonrpc_session_get_reconnect_stats(js, &rstats);
 612     status->state = rstats.state;
 613     status->sec_since_connect = rstats.msec_since_connect == UINT_MAX
 614         ? UINT_MAX : rstats.msec_since_connect / 1000;
 615     status->sec_since_disconnect = rstats.msec_since_disconnect == UINT_MAX
 616         ? UINT_MAX : rstats.msec_since_disconnect / 1000;
 617
 618     ds_init(&locks_held);
 619     ds_init(&locks_waiting);
 620     ds_init(&locks_lost);
 621     HMAP_FOR_EACH (waiter, session_node, &s->up.waiters) {
 622         struct ds *string;
 623
 624         string = (ovsdb_lock_waiter_is_owner(waiter) ? &locks_held
 625                   : waiter->mode == OVSDB_LOCK_WAIT ? &locks_waiting
 626                   : &locks_lost);
 627         if (string->length) {
 628             ds_put_char(string, ' ');
 629         }
 630         ds_put_cstr(string, waiter->lock_name);
 631     }
 632     status->locks_held = ds_steal_cstr(&locks_held);
 633     status->locks_waiting = ds_steal_cstr(&locks_waiting);
 634     status->locks_lost = ds_steal_cstr(&locks_lost);
 635
 636     status->n_connections = list_size(&remote->sessions);
 637
 638     return true;
 639 }
 640
 641 /* Examines 'request' to determine the database to which it relates, and then
 642  * searches 's' to find that database:
 643  *
 644  *    - If successful, returns the database and sets '*replyp' to NULL.
 645  *
 646  *    - If no such database exists, returns NULL and sets '*replyp' to an
 647  *      appropriate JSON-RPC error reply, owned by the caller. */
 648 static struct ovsdb *
 649 ovsdb_jsonrpc_lookup_db(const struct ovsdb_jsonrpc_session *s,
 650                         const struct jsonrpc_msg *request,
 651                         struct jsonrpc_msg **replyp)
 652 {
 653     struct json_array *params;
 654     struct ovsdb_error *error;
 655     const char *db_name;
 656     struct ovsdb *db;
 657
 658     params = json_array(request->params);
 659     if (!params->n || params->elems[0]->type != JSON_STRING) {
 660         error = ovsdb_syntax_error(
 661             request->params, NULL,
 662             "%s request params must begin with <db-name>", request->method);
 663         goto error;
 664     }
 665
 666     db_name = params->elems[0]->u.string;
 667     db = shash_find_data(&s->up.server->dbs, db_name);
 668     if (!db) {
 669         error = ovsdb_syntax_error(
 670             request->params, "unknown database",
 671             "%s request specifies unknown database %s",
 672             request->method, db_name);
 673         goto error;
 674     }
 675
 676     *replyp = NULL;
 677     return db;
 678
 679 error:
 680     *replyp = jsonrpc_create_error(ovsdb_error_to_json(error), request->id);
 681     ovsdb_error_destroy(error);
 682     return NULL;
 683 }
 684
 685 static struct ovsdb_error *
 686 ovsdb_jsonrpc_session_parse_lock_name(const struct jsonrpc_msg *request,
 687                                       const char **lock_namep)
 688 {
 689     const struct json_array *params;
 690
 691     params = json_array(request->params);
 692     if (params->n != 1 || params->elems[0]->type != JSON_STRING ||
 693         !ovsdb_parser_is_id(json_string(params->elems[0]))) {
 694         *lock_namep = NULL;
 695         return ovsdb_syntax_error(request->params, NULL,
 696                                   "%s request params must be <id>",
 697                                   request->method);
 698     }
 699
 700     *lock_namep = json_string(params->elems[0]);
 701     return NULL;
 702 }
 703
 704 static void
 705 ovsdb_jsonrpc_session_notify(struct ovsdb_session *session,
 706                              const char *lock_name,
 707                              const char *method)
 708 {
 709     struct ovsdb_jsonrpc_session *s;
 710     struct json *params;
 711
 712     s = CONTAINER_OF(session, struct ovsdb_jsonrpc_session, up);
 713     params = json_array_create_1(json_string_create(lock_name));
 714     ovsdb_jsonrpc_session_send(s, jsonrpc_create_notify(method, params));
 715 }
 716
 717 static struct jsonrpc_msg *
 718 ovsdb_jsonrpc_session_lock(struct ovsdb_jsonrpc_session *s,
 719                            struct jsonrpc_msg *request,
 720                            enum ovsdb_lock_mode mode)
 721 {
 722     struct ovsdb_lock_waiter *waiter;
 723     struct jsonrpc_msg *reply;
 724     struct ovsdb_error *error;
 725     struct ovsdb_session *victim;
 726     const char *lock_name;
 727     struct json *result;
 728
 729     error = ovsdb_jsonrpc_session_parse_lock_name(request, &lock_name);
 730     if (error) {
 731         goto error;
 732     }
 733
 734     /* Report error if this session has issued a "lock" or "steal" without a
 735      * matching "unlock" for this lock. */
 736     waiter = ovsdb_session_get_lock_waiter(&s->up, lock_name);
 737     if (waiter) {
 738         error = ovsdb_syntax_error(
 739             request->params, NULL,
 740             "must issue \"unlock\" before new \"%s\"", request->method);
 741         goto error;
 742     }
 743
 744     /* Get the lock, add us as a waiter. */
 745     waiter = ovsdb_server_lock(&s->remote->server->up, &s->up, lock_name, mode,
 746                                &victim);
 747     if (victim) {
 748         ovsdb_jsonrpc_session_notify(victim, lock_name, "stolen");
 749     }
 750
 751     result = json_object_create();
 752     json_object_put(result, "locked",
 753                     json_boolean_create(ovsdb_lock_waiter_is_owner(waiter)));
 754
 755     return jsonrpc_create_reply(result, request->id);
 756
 757 error:
 758     reply = jsonrpc_create_error(ovsdb_error_to_json(error), request->id);
 759     ovsdb_error_destroy(error);
 760     return reply;
 761 }
 762
 763 static void
 764 ovsdb_jsonrpc_session_unlock_all(struct ovsdb_jsonrpc_session *s)
 765 {
 766     struct ovsdb_lock_waiter *waiter, *next;
 767
 768     HMAP_FOR_EACH_SAFE (waiter, next, session_node, &s->up.waiters) {
 769         ovsdb_jsonrpc_session_unlock__(waiter);
 770     }
 771 }
 772
 773 static void
 774 ovsdb_jsonrpc_session_unlock__(struct ovsdb_lock_waiter *waiter)
 775 {
 776     struct ovsdb_lock *lock = waiter->lock;
 777
 778     if (lock) {
 779         struct ovsdb_session *new_owner = ovsdb_lock_waiter_remove(waiter);
 780         if (new_owner) {
 781             ovsdb_jsonrpc_session_notify(new_owner, lock->name, "locked");
 782         } else {
 783             /* ovsdb_server_lock() might have freed 'lock'. */
 784         }
 785     }
 786
 787     ovsdb_lock_waiter_destroy(waiter);
 788 }
 789
 790 static struct jsonrpc_msg *
 791 ovsdb_jsonrpc_session_unlock(struct ovsdb_jsonrpc_session *s,
 792                              struct jsonrpc_msg *request)
 793 {
 794     struct ovsdb_lock_waiter *waiter;
 795     struct jsonrpc_msg *reply;
 796     struct ovsdb_error *error;
 797     const char *lock_name;
 798
 799     error = ovsdb_jsonrpc_session_parse_lock_name(request, &lock_name);
 800     if (error) {
 801         goto error;
 802     }
 803
 804     /* Report error if this session has not issued a "lock" or "steal" for this
 805      * lock. */
 806     waiter = ovsdb_session_get_lock_waiter(&s->up, lock_name);
 807     if (!waiter) {
 808         error = ovsdb_syntax_error(
 809             request->params, NULL, "\"unlock\" without \"lock\" or \"steal\"");
 810         goto error;
 811     }
 812
 813     ovsdb_jsonrpc_session_unlock__(waiter);
 814
 815     return jsonrpc_create_reply(json_object_create(), request->id);
 816
 817 error:
 818     reply = jsonrpc_create_error(ovsdb_error_to_json(error), request->id);
 819     ovsdb_error_destroy(error);
 820     return reply;
 821 }
 822
 823 static struct jsonrpc_msg *
 824 execute_transaction(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
 825                     struct jsonrpc_msg *request)
 826 {
 827     ovsdb_jsonrpc_trigger_create(s, db, request->id, request->params);
 828     request->id = NULL;
 829     request->params = NULL;
 830     jsonrpc_msg_destroy(request);
 831     return NULL;
 832 }
 833
 834 static void
 835 ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *s,
 836                                   struct jsonrpc_msg *request)
 837 {
 838     struct jsonrpc_msg *reply;
 839
 840     if (!strcmp(request->method, "transact")) {
 841         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
 842         if (!reply) {
 843             reply = execute_transaction(s, db, request);
 844         }
 845     } else if (!strcmp(request->method, "monitor")) {
 846         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
 847         if (!reply) {
 848             reply = ovsdb_jsonrpc_monitor_create(s, db, request->params,
 849                                                  request->id);
 850         }
 851     } else if (!strcmp(request->method, "monitor_cancel")) {
 852         reply = ovsdb_jsonrpc_monitor_cancel(s, json_array(request->params),
 853                                              request->id);
 854     } else if (!strcmp(request->method, "get_schema")) {
 855         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
 856         if (!reply) {
 857             reply = jsonrpc_create_reply(ovsdb_schema_to_json(db->schema),
 858                                          request->id);
 859         }
 860     } else if (!strcmp(request->method, "list_dbs")) {
 861         size_t n_dbs = shash_count(&s->up.server->dbs);
 862         struct shash_node *node;
 863         struct json **dbs;
 864         size_t i;
 865
 866         dbs = xmalloc(n_dbs * sizeof *dbs);
 867         i = 0;
 868         SHASH_FOR_EACH (node, &s->up.server->dbs) {
 869             dbs[i++] = json_string_create(node->name);
 870         }
 871         reply = jsonrpc_create_reply(json_array_create(dbs, n_dbs),
 872                                      request->id);
 873     } else if (!strcmp(request->method, "lock")) {
 874         reply = ovsdb_jsonrpc_session_lock(s, request, OVSDB_LOCK_WAIT);
 875     } else if (!strcmp(request->method, "steal")) {
 876         reply = ovsdb_jsonrpc_session_lock(s, request, OVSDB_LOCK_STEAL);
 877     } else if (!strcmp(request->method, "unlock")) {
 878         reply = ovsdb_jsonrpc_session_unlock(s, request);
 879     } else if (!strcmp(request->method, "echo")) {
 880         reply = jsonrpc_create_reply(json_clone(request->params), request->id);
 881     } else {
 882         reply = jsonrpc_create_error(json_string_create("unknown method"),
 883                                      request->id);
 884     }
 885
 886     if (reply) {
 887         jsonrpc_msg_destroy(request);
 888         ovsdb_jsonrpc_session_send(s, reply);
 889     }
 890 }
 891
 892 static void
 893 execute_cancel(struct ovsdb_jsonrpc_session *s, struct jsonrpc_msg *request)
 894 {
 895     if (json_array(request->params)->n == 1) {
 896         struct ovsdb_jsonrpc_trigger *t;
 897         struct json *id;
 898
 899         id = request->params->u.array.elems[0];
 900         t = ovsdb_jsonrpc_trigger_find(s, id, json_hash(id, 0));
 901         if (t) {
 902             ovsdb_jsonrpc_trigger_complete(t);
 903         }
 904     }
 905 }
 906
 907 static void
 908 ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *s,
 909                                  struct jsonrpc_msg *request)
 910 {
 911     if (!strcmp(request->method, "cancel")) {
 912         execute_cancel(s, request);
 913     }
 914     jsonrpc_msg_destroy(request);
 915 }
 916
 917 static void
 918 ovsdb_jsonrpc_session_send(struct ovsdb_jsonrpc_session *s,
 919                            struct jsonrpc_msg *msg)
 920 {
 921     ovsdb_jsonrpc_monitor_flush_all(s);
 922     jsonrpc_session_send(s->js, msg);
 923 }
 924 \f
 925 /* JSON-RPC database server triggers.
 926  *
 927  * (Every transaction is treated as a trigger even if it doesn't actually have
 928  * any "wait" operations.) */
 929
 930 struct ovsdb_jsonrpc_trigger {
 931     struct ovsdb_trigger trigger;
 932     struct hmap_node hmap_node; /* In session's "triggers" hmap. */
 933     struct json *id;
 934 };
 935
 936 static void
 937 ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
 938                              struct json *id, struct json *params)
 939 {
 940     struct ovsdb_jsonrpc_trigger *t;
 941     size_t hash;
 942
 943     /* Check for duplicate ID. */
 944     hash = json_hash(id, 0);
 945     t = ovsdb_jsonrpc_trigger_find(s, id, hash);
 946     if (t) {
 947         struct jsonrpc_msg *msg;
 948
 949         msg = jsonrpc_create_error(json_string_create("duplicate request ID"),
 950                                    id);
 951         ovsdb_jsonrpc_session_send(s, msg);
 952         json_destroy(id);
 953         json_destroy(params);
 954         return;
 955     }
 956
 957     /* Insert into trigger table. */
 958     t = xmalloc(sizeof *t);
 959     ovsdb_trigger_init(&s->up, db, &t->trigger, params, time_msec());
 960     t->id = id;
 961     hmap_insert(&s->triggers, &t->hmap_node, hash);
 962
 963     /* Complete early if possible. */
 964     if (ovsdb_trigger_is_complete(&t->trigger)) {
 965         ovsdb_jsonrpc_trigger_complete(t);
 966     }
 967 }
 968
 969 static struct ovsdb_jsonrpc_trigger *
 970 ovsdb_jsonrpc_trigger_find(struct ovsdb_jsonrpc_session *s,
 971                            const struct json *id, size_t hash)
 972 {
 973     struct ovsdb_jsonrpc_trigger *t;
 974
 975     HMAP_FOR_EACH_WITH_HASH (t, hmap_node, hash, &s->triggers) {
 976         if (json_equal(t->id, id)) {
 977             return t;
 978         }
 979     }
 980
 981     return NULL;
 982 }
 983
 984 static void
 985 ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *t)
 986 {
 987     struct ovsdb_jsonrpc_session *s;
 988
 989     s = CONTAINER_OF(t->trigger.session, struct ovsdb_jsonrpc_session, up);
 990
 991     if (jsonrpc_session_is_connected(s->js)) {
 992         struct jsonrpc_msg *reply;
 993         struct json *result;
 994
 995         result = ovsdb_trigger_steal_result(&t->trigger);
 996         if (result) {
 997             reply = jsonrpc_create_reply(result, t->id);
 998         } else {
 999             reply = jsonrpc_create_error(json_string_create("canceled"),
1000                                          t->id);
1001         }
1002         ovsdb_jsonrpc_session_send(s, reply);
1003     }
1004
1005     json_destroy(t->id);
1006     ovsdb_trigger_destroy(&t->trigger);
1007     hmap_remove(&s->triggers, &t->hmap_node);
1008     free(t);
1009 }
1010
1011 static void
1012 ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *s)
1013 {
1014     struct ovsdb_jsonrpc_trigger *t, *next;
1015     HMAP_FOR_EACH_SAFE (t, next, hmap_node, &s->triggers) {
1016         ovsdb_jsonrpc_trigger_complete(t);
1017     }
1018 }
1019
1020 static void
1021 ovsdb_jsonrpc_trigger_complete_done(struct ovsdb_jsonrpc_session *s)
1022 {
1023     while (!list_is_empty(&s->up.completions)) {
1024         struct ovsdb_jsonrpc_trigger *t
1025             = CONTAINER_OF(s->up.completions.next,
1026                            struct ovsdb_jsonrpc_trigger, trigger.node);
1027         ovsdb_jsonrpc_trigger_complete(t);
1028     }
1029 }
1030 \f
1031 /* database table monitors. */
1032
1033 enum ovsdb_monitor_selection {
1034     OJMS_INITIAL = 1 << 0,      /* All rows when monitor is created. */
1035     OJMS_INSERT = 1 << 1,       /* New rows. */
1036     OJMS_DELETE = 1 << 2,       /* Deleted rows. */
1037     OJMS_MODIFY = 1 << 3        /* Modified rows. */
1038 };
1039
1040 /* A particular column being monitored. */
1041 struct ovsdb_monitor_column {
1042     const struct ovsdb_column *column;
1043     enum ovsdb_monitor_selection select;
1044 };
1045
1046 /* A row that has changed in a monitored table. */
1047 struct ovsdb_monitor_row {
1048     struct hmap_node hmap_node; /* In ovsdb_jsonrpc_monitor_table.changes. */
1049     struct uuid uuid;           /* UUID of row that changed. */
1050     struct ovsdb_datum *old;    /* Old data, NULL for an inserted row. */
1051     struct ovsdb_datum *new;    /* New data, NULL for a deleted row. */
1052 };
1053
1054 /* A particular table being monitored. */
1055 struct ovsdb_monitor_table {
1056     const struct ovsdb_table *table;
1057
1058     /* This is the union (bitwise-OR) of the 'select' values in all of the
1059      * members of 'columns' below. */
1060     enum ovsdb_monitor_selection select;
1061
1062     /* Columns being monitored. */
1063     struct ovsdb_monitor_column *columns;
1064     size_t n_columns;
1065
1066     /* Contains 'struct ovsdb_monitor_row's for rows that have been
1067      * updated but not yet flushed to the jsonrpc connection. */
1068     struct hmap changes;
1069 };
1070
1071 struct ovsdb_jsonrpc_monitor;
1072 /*  Backend monitor.
1073  *
1074  *  ovsdb_monitor keep track of the ovsdb changes.
1075  */
1076 /* A collection of tables being monitored. */
1077 struct ovsdb_monitor {
1078     struct ovsdb_replica replica;
1079     struct shash tables;     /* Holds "struct ovsdb_monitor_table"s. */
1080     struct ovsdb_jsonrpc_monitor *jsonrpc_monitor;
1081 };
1082
1083 /* Jsonrpc front end monitor. */
1084 struct ovsdb_jsonrpc_monitor {
1085     struct ovsdb_jsonrpc_session *session;
1086     struct ovsdb *db;
1087     struct hmap_node node;      /* In ovsdb_jsonrpc_session's "monitors". */
1088
1089     struct json *monitor_id;
1090     struct ovsdb_monitor *dbmon;
1091 };
1092
1093 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class;
1094
1095 struct ovsdb_jsonrpc_monitor *ovsdb_jsonrpc_monitor_find(
1096     struct ovsdb_jsonrpc_session *, const struct json *monitor_id);
1097 static void ovsdb_monitor_destroy(struct ovsdb_monitor *);
1098 static struct json *ovsdb_monitor_get_initial(
1099     const struct ovsdb_monitor *);
1100
1101 static bool
1102 parse_bool(struct ovsdb_parser *parser, const char *name, bool default_value)
1103 {
1104     const struct json *json;
1105
1106     json = ovsdb_parser_member(parser, name, OP_BOOLEAN | OP_OPTIONAL);
1107     return json ? json_boolean(json) : default_value;
1108 }
1109
1110 struct ovsdb_jsonrpc_monitor *
1111 ovsdb_jsonrpc_monitor_find(struct ovsdb_jsonrpc_session *s,
1112                            const struct json *monitor_id)
1113 {
1114     struct ovsdb_jsonrpc_monitor *m;
1115
1116     HMAP_FOR_EACH_WITH_HASH (m, node, json_hash(monitor_id, 0), &s->monitors) {
1117         if (json_equal(m->monitor_id, monitor_id)) {
1118             return m;
1119         }
1120     }
1121
1122     return NULL;
1123 }
1124
1125 static void
1126 ovsdb_monitor_add_column(struct ovsdb_monitor *dbmon,
1127                          const struct ovsdb_table *table,
1128                          const struct ovsdb_column *column,
1129                          enum ovsdb_monitor_selection select,
1130                          size_t *allocated_columns)
1131 {
1132     struct ovsdb_monitor_table *mt;
1133     struct ovsdb_monitor_column *c;
1134
1135     mt = shash_find_data(&dbmon->tables, table->schema->name);
1136
1137     if (mt->n_columns >= *allocated_columns) {
1138         mt->columns = x2nrealloc(mt->columns, allocated_columns,
1139                                  sizeof *mt->columns);
1140     }
1141
1142     c = &mt->columns[mt->n_columns++];
1143     c->column = column;
1144     c->select = select;
1145 }
1146
1147 static int
1148 compare_ovsdb_monitor_column(const void *a_, const void *b_)
1149 {
1150     const struct ovsdb_monitor_column *a = a_;
1151     const struct ovsdb_monitor_column *b = b_;
1152
1153     return a->column < b->column ? -1 : a->column > b->column;
1154 }
1155
1156 static void
1157 ovsdb_monitor_table_add_select(struct ovsdb_monitor *dbmon,
1158                                const struct ovsdb_table *table,
1159                                enum ovsdb_monitor_selection select)
1160 {
1161     struct ovsdb_monitor_table * mt;
1162     mt = shash_find_data(&dbmon->tables, table->schema->name);
1163     mt->select |= select;
1164 }
1165
1166 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
1167 ovsdb_jsonrpc_parse_monitor_request(struct ovsdb_monitor *dbmon,
1168                                     const struct ovsdb_table *table,
1169                                     const struct json *monitor_request,
1170                                     size_t *allocated_columns)
1171 {
1172     const struct ovsdb_table_schema *ts = table->schema;
1173     enum ovsdb_monitor_selection select;
1174     const struct json *columns, *select_json;
1175     struct ovsdb_parser parser;
1176     struct ovsdb_error *error;
1177
1178     ovsdb_parser_init(&parser, monitor_request, "table %s", ts->name);
1179     columns = ovsdb_parser_member(&parser, "columns", OP_ARRAY | OP_OPTIONAL);
1180     select_json = ovsdb_parser_member(&parser, "select",
1181                                       OP_OBJECT | OP_OPTIONAL);
1182     error = ovsdb_parser_finish(&parser);
1183     if (error) {
1184         return error;
1185     }
1186
1187     if (select_json) {
1188         select = 0;
1189         ovsdb_parser_init(&parser, select_json, "table %s select", ts->name);
1190         if (parse_bool(&parser, "initial", true)) {
1191             select |= OJMS_INITIAL;
1192         }
1193         if (parse_bool(&parser, "insert", true)) {
1194             select |= OJMS_INSERT;
1195         }
1196         if (parse_bool(&parser, "delete", true)) {
1197             select |= OJMS_DELETE;
1198         }
1199         if (parse_bool(&parser, "modify", true)) {
1200             select |= OJMS_MODIFY;
1201         }
1202         error = ovsdb_parser_finish(&parser);
1203         if (error) {
1204             return error;
1205         }
1206     } else {
1207         select = OJMS_INITIAL | OJMS_INSERT | OJMS_DELETE | OJMS_MODIFY;
1208     }
1209
1210     ovsdb_monitor_table_add_select(dbmon, table, select);
1211     if (columns) {
1212         size_t i;
1213
1214         if (columns->type != JSON_ARRAY) {
1215             return ovsdb_syntax_error(columns, NULL,
1216                                       "array of column names expected");
1217         }
1218
1219         for (i = 0; i < columns->u.array.n; i++) {
1220             const struct ovsdb_column *column;
1221             const char *s;
1222
1223             if (columns->u.array.elems[i]->type != JSON_STRING) {
1224                 return ovsdb_syntax_error(columns, NULL,
1225                                           "array of column names expected");
1226             }
1227
1228             s = columns->u.array.elems[i]->u.string;
1229             column = shash_find_data(&table->schema->columns, s);
1230             if (!column) {
1231                 return ovsdb_syntax_error(columns, NULL, "%s is not a valid "
1232                                           "column name", s);
1233             }
1234             ovsdb_monitor_add_column(dbmon, table, column, select,
1235                                      allocated_columns);
1236         }
1237     } else {
1238         struct shash_node *node;
1239
1240         SHASH_FOR_EACH (node, &ts->columns) {
1241             const struct ovsdb_column *column = node->data;
1242             if (column->index != OVSDB_COL_UUID) {
1243                 ovsdb_monitor_add_column(dbmon, table, column, select,
1244                                          allocated_columns);
1245             }
1246         }
1247     }
1248
1249     return NULL;
1250 }
1251
1252 static struct ovsdb_monitor *
1253 ovsdb_monitor_create(struct ovsdb *db,
1254                      struct ovsdb_jsonrpc_monitor *jsonrpc_monitor,
1255                      const struct ovsdb_replica_class *replica_class)
1256 {
1257     struct ovsdb_monitor *dbmon;
1258
1259     dbmon = xzalloc(sizeof *dbmon);
1260
1261     ovsdb_replica_init(&dbmon->replica, replica_class);
1262     ovsdb_add_replica(db, &dbmon->replica);
1263     dbmon->jsonrpc_monitor = jsonrpc_monitor;
1264     shash_init(&dbmon->tables);
1265
1266     return dbmon;
1267 }
1268
1269 static void
1270 ovsdb_monitor_add_table(struct ovsdb_monitor *m,
1271                         const struct ovsdb_table *table)
1272 {
1273     struct ovsdb_monitor_table *mt;
1274
1275     mt = xzalloc(sizeof *mt);
1276     mt->table = table;
1277     hmap_init(&mt->changes);
1278     shash_add(&m->tables, table->schema->name, mt);
1279 }
1280
1281 /* Check for duplicated column names. Return the first
1282  * duplicated column's name if found. Otherwise return
1283  * NULL.  */
1284 static const char * OVS_WARN_UNUSED_RESULT
1285 ovsdb_monitor_table_check_duplicates(struct ovsdb_monitor *m,
1286                           const struct ovsdb_table *table)
1287 {
1288     struct ovsdb_monitor_table *mt;
1289     int i;
1290
1291     mt = shash_find_data(&m->tables, table->schema->name);
1292
1293     if (mt) {
1294         /* Check for duplicate columns. */
1295         qsort(mt->columns, mt->n_columns, sizeof *mt->columns,
1296               compare_ovsdb_monitor_column);
1297         for (i = 1; i < mt->n_columns; i++) {
1298             if (mt->columns[i].column == mt->columns[i - 1].column) {
1299                 return mt->columns[i].column->name;
1300             }
1301         }
1302     }
1303
1304     return NULL;
1305 }
1306
1307 static struct jsonrpc_msg *
1308 ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
1309                              struct json *params,
1310                              const struct json *request_id)
1311 {
1312     struct ovsdb_jsonrpc_monitor *m = NULL;
1313     struct json *monitor_id, *monitor_requests;
1314     struct ovsdb_error *error = NULL;
1315     struct shash_node *node;
1316     struct json *json;
1317
1318     if (json_array(params)->n != 3) {
1319         error = ovsdb_syntax_error(params, NULL, "invalid parameters");
1320         goto error;
1321     }
1322     monitor_id = params->u.array.elems[1];
1323     monitor_requests = params->u.array.elems[2];
1324     if (monitor_requests->type != JSON_OBJECT) {
1325         error = ovsdb_syntax_error(monitor_requests, NULL,
1326                                    "monitor-requests must be object");
1327         goto error;
1328     }
1329
1330     if (ovsdb_jsonrpc_monitor_find(s, monitor_id)) {
1331         error = ovsdb_syntax_error(monitor_id, NULL, "duplicate monitor ID");
1332         goto error;
1333     }
1334
1335     m = xzalloc(sizeof *m);
1336     m->session = s;
1337     m->db = db;
1338     m->dbmon = ovsdb_monitor_create(db, m, &ovsdb_jsonrpc_replica_class);
1339     hmap_insert(&s->monitors, &m->node, json_hash(monitor_id, 0));
1340     m->monitor_id = json_clone(monitor_id);
1341
1342     SHASH_FOR_EACH (node, json_object(monitor_requests)) {
1343         const struct ovsdb_table *table;
1344         const char *column_name;
1345         size_t allocated_columns;
1346         const struct json *mr_value;
1347         size_t i;
1348
1349         table = ovsdb_get_table(m->db, node->name);
1350         if (!table) {
1351             error = ovsdb_syntax_error(NULL, NULL,
1352                                        "no table named %s", node->name);
1353             goto error;
1354         }
1355
1356         ovsdb_monitor_add_table(m->dbmon, table);
1357
1358         /* Parse columns. */
1359         mr_value = node->data;
1360         allocated_columns = 0;
1361         if (mr_value->type == JSON_ARRAY) {
1362             const struct json_array *array = &mr_value->u.array;
1363
1364             for (i = 0; i < array->n; i++) {
1365                 error = ovsdb_jsonrpc_parse_monitor_request(
1366                     m->dbmon, table, array->elems[i], &allocated_columns);
1367                 if (error) {
1368                     goto error;
1369                 }
1370             }
1371         } else {
1372             error = ovsdb_jsonrpc_parse_monitor_request(
1373                 m->dbmon, table, mr_value, &allocated_columns);
1374             if (error) {
1375                 goto error;
1376             }
1377         }
1378
1379         column_name = ovsdb_monitor_table_check_duplicates(m->dbmon, table);
1380
1381         if (column_name) {
1382             error = ovsdb_syntax_error(mr_value, NULL, "column %s "
1383                                        "mentioned more than once",
1384                                         column_name);
1385             goto error;
1386         }
1387     }
1388
1389     return jsonrpc_create_reply(ovsdb_monitor_get_initial(m->dbmon),
1390                                 request_id);
1391
1392 error:
1393     if (m) {
1394         ovsdb_jsonrpc_monitor_destroy(m);
1395     }
1396
1397     json = ovsdb_error_to_json(error);
1398     ovsdb_error_destroy(error);
1399     return jsonrpc_create_error(json, request_id);
1400 }
1401
1402 static struct jsonrpc_msg *
1403 ovsdb_jsonrpc_monitor_cancel(struct ovsdb_jsonrpc_session *s,
1404                              struct json_array *params,
1405                              const struct json *request_id)
1406 {
1407     if (params->n != 1) {
1408         return jsonrpc_create_error(json_string_create("invalid parameters"),
1409                                     request_id);
1410     } else {
1411         struct ovsdb_jsonrpc_monitor *m;
1412
1413         m = ovsdb_jsonrpc_monitor_find(s, params->elems[0]);
1414         if (!m) {
1415             return jsonrpc_create_error(json_string_create("unknown monitor"),
1416                                         request_id);
1417         } else {
1418             ovsdb_jsonrpc_monitor_destroy(m);
1419             return jsonrpc_create_reply(json_object_create(), request_id);
1420         }
1421     }
1422 }
1423
1424 static void
1425 ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *s)
1426 {
1427     struct ovsdb_jsonrpc_monitor *m, *next;
1428
1429     HMAP_FOR_EACH_SAFE (m, next, node, &s->monitors) {
1430         ovsdb_jsonrpc_monitor_destroy(m);
1431     }
1432 }
1433
1434 static struct ovsdb_monitor *
1435 ovsdb_monitor_cast(struct ovsdb_replica *replica)
1436 {
1437     ovs_assert(replica->class == &ovsdb_jsonrpc_replica_class);
1438     return CONTAINER_OF(replica, struct ovsdb_monitor, replica);
1439 }
1440
1441 struct ovsdb_monitor_aux {
1442     const struct ovsdb_monitor *monitor;
1443     struct ovsdb_monitor_table *mt;
1444 };
1445
1446 /* Finds and returns the ovsdb_monitor_row in 'mt->changes' for the
1447  * given 'uuid', or NULL if there is no such row. */
1448 static struct ovsdb_monitor_row *
1449 ovsdb_monitor_row_find(const struct ovsdb_monitor_table *mt,
1450                        const struct uuid *uuid)
1451 {
1452     struct ovsdb_monitor_row *row;
1453
1454     HMAP_FOR_EACH_WITH_HASH (row, hmap_node, uuid_hash(uuid), &mt->changes) {
1455         if (uuid_equals(uuid, &row->uuid)) {
1456             return row;
1457         }
1458     }
1459     return NULL;
1460 }
1461
1462 /* Allocates an array of 'mt->n_columns' ovsdb_datums and initializes them as
1463  * copies of the data in 'row' drawn from the columns represented by
1464  * mt->columns[].  Returns the array.
1465  *
1466  * If 'row' is NULL, returns NULL. */
1467 static struct ovsdb_datum *
1468 clone_monitor_row_data(const struct ovsdb_monitor_table *mt,
1469                        const struct ovsdb_row *row)
1470 {
1471     struct ovsdb_datum *data;
1472     size_t i;
1473
1474     if (!row) {
1475         return NULL;
1476     }
1477
1478     data = xmalloc(mt->n_columns * sizeof *data);
1479     for (i = 0; i < mt->n_columns; i++) {
1480         const struct ovsdb_column *c = mt->columns[i].column;
1481         const struct ovsdb_datum *src = &row->fields[c->index];
1482         struct ovsdb_datum *dst = &data[i];
1483         const struct ovsdb_type *type = &c->type;
1484
1485         ovsdb_datum_clone(dst, src, type);
1486     }
1487     return data;
1488 }
1489
1490 /* Replaces the mt->n_columns ovsdb_datums in row[] by copies of the data from
1491  * in 'row' drawn from the columns represented by mt->columns[]. */
1492 static void
1493 update_monitor_row_data(const struct ovsdb_monitor_table *mt,
1494                         const struct ovsdb_row *row,
1495                         struct ovsdb_datum *data)
1496 {
1497     size_t i;
1498
1499     for (i = 0; i < mt->n_columns; i++) {
1500         const struct ovsdb_column *c = mt->columns[i].column;
1501         const struct ovsdb_datum *src = &row->fields[c->index];
1502         struct ovsdb_datum *dst = &data[i];
1503         const struct ovsdb_type *type = &c->type;
1504
1505         if (!ovsdb_datum_equals(src, dst, type)) {
1506             ovsdb_datum_destroy(dst, type);
1507             ovsdb_datum_clone(dst, src, type);
1508         }
1509     }
1510 }
1511
1512 /* Frees all of the mt->n_columns ovsdb_datums in data[], using the types taken
1513  * from mt->columns[], plus 'data' itself. */
1514 static void
1515 free_monitor_row_data(const struct ovsdb_monitor_table *mt,
1516                       struct ovsdb_datum *data)
1517 {
1518     if (data) {
1519         size_t i;
1520
1521         for (i = 0; i < mt->n_columns; i++) {
1522             const struct ovsdb_column *c = mt->columns[i].column;
1523
1524             ovsdb_datum_destroy(&data[i], &c->type);
1525         }
1526         free(data);
1527     }
1528 }
1529
1530 /* Frees 'row', which must have been created from 'mt'. */
1531 static void
1532 ovsdb_monitor_row_destroy(const struct ovsdb_monitor_table *mt,
1533                           struct ovsdb_monitor_row *row)
1534 {
1535     if (row) {
1536         free_monitor_row_data(mt, row->old);
1537         free_monitor_row_data(mt, row->new);
1538         free(row);
1539     }
1540 }
1541
1542 static bool
1543 ovsdb_monitor_change_cb(const struct ovsdb_row *old,
1544                         const struct ovsdb_row *new,
1545                         const unsigned long int *changed OVS_UNUSED,
1546                         void *aux_)
1547 {
1548     struct ovsdb_monitor_aux *aux = aux_;
1549     const struct ovsdb_monitor *m = aux->monitor;
1550     struct ovsdb_table *table = new ? new->table : old->table;
1551     const struct uuid *uuid = ovsdb_row_get_uuid(new ? new : old);
1552     struct ovsdb_monitor_row *change;
1553     struct ovsdb_monitor_table *mt;
1554
1555     if (!aux->mt || table != aux->mt->table) {
1556         aux->mt = shash_find_data(&m->tables, table->schema->name);
1557         if (!aux->mt) {
1558             /* We don't care about rows in this table at all.  Tell the caller
1559              * to skip it.  */
1560             return false;
1561         }
1562     }
1563     mt = aux->mt;
1564
1565     change = ovsdb_monitor_row_find(mt, uuid);
1566     if (!change) {
1567         change = xmalloc(sizeof *change);
1568         hmap_insert(&mt->changes, &change->hmap_node, uuid_hash(uuid));
1569         change->uuid = *uuid;
1570         change->old = clone_monitor_row_data(mt, old);
1571         change->new = clone_monitor_row_data(mt, new);
1572     } else {
1573         if (new) {
1574             update_monitor_row_data(mt, new, change->new);
1575         } else {
1576             free_monitor_row_data(mt, change->new);
1577             change->new = NULL;
1578
1579             if (!change->old) {
1580                 /* This row was added then deleted.  Forget about it. */
1581                 hmap_remove(&mt->changes, &change->hmap_node);
1582                 free(change);
1583             }
1584         }
1585     }
1586     return true;
1587 }
1588
1589 /* Returns JSON for a <row-update> (as described in RFC 7047) for 'row' within
1590  * 'mt', or NULL if no row update should be sent.
1591  *
1592  * The caller should specify 'initial' as true if the returned JSON is going to
1593  * be used as part of the initial reply to a "monitor" request, false if it is
1594  * going to be used as part of an "update" notification.
1595  *
1596  * 'changed' must be a scratch buffer for internal use that is at least
1597  * bitmap_n_bytes(mt->n_columns) bytes long. */
1598 static struct json *
1599 ovsdb_monitor_compose_row_update(
1600     const struct ovsdb_monitor_table *mt,
1601     const struct ovsdb_monitor_row *row,
1602     bool initial, unsigned long int *changed)
1603 {
1604     enum ovsdb_monitor_selection type;
1605     struct json *old_json, *new_json;
1606     struct json *row_json;
1607     size_t i;
1608
1609     type = (initial ? OJMS_INITIAL
1610             : !row->old ? OJMS_INSERT
1611             : !row->new ? OJMS_DELETE
1612             : OJMS_MODIFY);
1613     if (!(mt->select & type)) {
1614         return NULL;
1615     }
1616
1617     if (type == OJMS_MODIFY) {
1618         size_t n_changes;
1619
1620         n_changes = 0;
1621         memset(changed, 0, bitmap_n_bytes(mt->n_columns));
1622         for (i = 0; i < mt->n_columns; i++) {
1623             const struct ovsdb_column *c = mt->columns[i].column;
1624             if (!ovsdb_datum_equals(&row->old[i], &row->new[i], &c->type)) {
1625                 bitmap_set1(changed, i);
1626                 n_changes++;
1627             }
1628         }
1629         if (!n_changes) {
1630             /* No actual changes: presumably a row changed and then
1631              * changed back later. */
1632             return NULL;
1633         }
1634     }
1635
1636     row_json = json_object_create();
1637     old_json = new_json = NULL;
1638     if (type & (OJMS_DELETE | OJMS_MODIFY)) {
1639         old_json = json_object_create();
1640         json_object_put(row_json, "old", old_json);
1641     }
1642     if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
1643         new_json = json_object_create();
1644         json_object_put(row_json, "new", new_json);
1645     }
1646     for (i = 0; i < mt->n_columns; i++) {
1647         const struct ovsdb_monitor_column *c = &mt->columns[i];
1648
1649         if (!(type & c->select)) {
1650             /* We don't care about this type of change for this
1651              * particular column (but we will care about it for some
1652              * other column). */
1653             continue;
1654         }
1655
1656         if ((type == OJMS_MODIFY && bitmap_is_set(changed, i))
1657             || type == OJMS_DELETE) {
1658             json_object_put(old_json, c->column->name,
1659                             ovsdb_datum_to_json(&row->old[i],
1660                                                 &c->column->type));
1661         }
1662         if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
1663             json_object_put(new_json, c->column->name,
1664                             ovsdb_datum_to_json(&row->new[i],
1665                                                 &c->column->type));
1666         }
1667     }
1668
1669     return row_json;
1670 }
1671
1672 /* Constructs and returns JSON for a <table-updates> object (as described in
1673  * RFC 7047) for all the outstanding changes within 'monitor', and deletes all
1674  * the outstanding changes from 'monitor'.  Returns NULL if no update needs to
1675  * be sent.
1676  *
1677  * The caller should specify 'initial' as true if the returned JSON is going to
1678  * be used as part of the initial reply to a "monitor" request, false if it is
1679  * going to be used as part of an "update" notification. */
1680 static struct json *
1681 ovsdb_monitor_compose_table_update(
1682     const struct ovsdb_monitor *dbmon, bool initial)
1683 {
1684     struct shash_node *node;
1685     unsigned long int *changed;
1686     struct json *json;
1687     size_t max_columns;
1688
1689     max_columns = 0;
1690     SHASH_FOR_EACH (node, &dbmon->tables) {
1691         struct ovsdb_monitor_table *mt = node->data;
1692
1693         max_columns = MAX(max_columns, mt->n_columns);
1694     }
1695     changed = xmalloc(bitmap_n_bytes(max_columns));
1696
1697     json = NULL;
1698     SHASH_FOR_EACH (node, &dbmon->tables) {
1699         struct ovsdb_monitor_table *mt = node->data;
1700         struct ovsdb_monitor_row *row, *next;
1701         struct json *table_json = NULL;
1702
1703         HMAP_FOR_EACH_SAFE (row, next, hmap_node, &mt->changes) {
1704             struct json *row_json;
1705
1706             row_json = ovsdb_monitor_compose_row_update(
1707                 mt, row, initial, changed);
1708             if (row_json) {
1709                 char uuid[UUID_LEN + 1];
1710
1711                 /* Create JSON object for transaction overall. */
1712                 if (!json) {
1713                     json = json_object_create();
1714                 }
1715
1716                 /* Create JSON object for transaction on this table. */
1717                 if (!table_json) {
1718                     table_json = json_object_create();
1719                     json_object_put(json, mt->table->schema->name, table_json);
1720                 }
1721
1722                 /* Add JSON row to JSON table. */
1723                 snprintf(uuid, sizeof uuid, UUID_FMT, UUID_ARGS(&row->uuid));
1724                 json_object_put(table_json, uuid, row_json);
1725             }
1726
1727             hmap_remove(&mt->changes, &row->hmap_node);
1728             ovsdb_monitor_row_destroy(mt, row);
1729         }
1730     }
1731
1732     free(changed);
1733
1734     return json;
1735 }
1736
1737 static struct json *
1738 ovsdb_jsonrpc_monitor_compose_table_update(
1739     const struct ovsdb_jsonrpc_monitor *monitor, bool initial)
1740 {
1741     return ovsdb_monitor_compose_table_update(monitor->dbmon, initial);
1742 }
1743
1744 static bool
1745 ovsdb_monitor_needs_flush(struct ovsdb_monitor *dbmon)
1746 {
1747     struct shash_node *node;
1748
1749     SHASH_FOR_EACH (node, &dbmon->tables) {
1750         struct ovsdb_monitor_table *mt = node->data;
1751
1752         if (!hmap_is_empty(&mt->changes)) {
1753             return true;
1754         }
1755     }
1756     return false;
1757 }
1758
1759 static bool
1760 ovsdb_jsonrpc_monitor_needs_flush(struct ovsdb_jsonrpc_session *s)
1761 {
1762     struct ovsdb_jsonrpc_monitor *m;
1763
1764     HMAP_FOR_EACH (m, node, &s->monitors) {
1765         if (ovsdb_monitor_needs_flush(m->dbmon)) {
1766             return true;
1767         }
1768     }
1769
1770     return false;
1771 }
1772
1773 static void
1774 ovsdb_jsonrpc_monitor_flush_all(struct ovsdb_jsonrpc_session *s)
1775 {
1776     struct ovsdb_jsonrpc_monitor *m;
1777
1778     HMAP_FOR_EACH (m, node, &s->monitors) {
1779         struct json *json;
1780
1781         json = ovsdb_jsonrpc_monitor_compose_table_update(m, false);
1782         if (json) {
1783             struct jsonrpc_msg *msg;
1784             struct json *params;
1785
1786             params = json_array_create_2(json_clone(m->monitor_id), json);
1787             msg = jsonrpc_create_notify("update", params);
1788             jsonrpc_session_send(s->js, msg);
1789         }
1790     }
1791 }
1792
1793 static void
1794 ovsdb_monitor_init_aux(struct ovsdb_monitor_aux *aux,
1795                        const struct ovsdb_monitor *m)
1796 {
1797     aux->monitor = m;
1798     aux->mt = NULL;
1799 }
1800
1801 static struct ovsdb_error *
1802 ovsdb_monitor_commit(struct ovsdb_replica *replica,
1803                      const struct ovsdb_txn *txn,
1804                      bool durable OVS_UNUSED)
1805 {
1806     struct ovsdb_monitor *m = ovsdb_monitor_cast(replica);
1807     struct ovsdb_monitor_aux aux;
1808
1809     ovsdb_monitor_init_aux(&aux, m);
1810     ovsdb_txn_for_each_change(txn, ovsdb_monitor_change_cb, &aux);
1811
1812     return NULL;
1813 }
1814
1815 static struct json *
1816 ovsdb_monitor_get_initial(const struct ovsdb_monitor *dbmon)
1817 {
1818     struct ovsdb_monitor_aux aux;
1819     struct shash_node *node;
1820     struct json *json;
1821
1822     ovsdb_monitor_init_aux(&aux, dbmon);
1823     SHASH_FOR_EACH (node, &dbmon->tables) {
1824         struct ovsdb_monitor_table *mt = node->data;
1825
1826         if (mt->select & OJMS_INITIAL) {
1827             struct ovsdb_row *row;
1828
1829             HMAP_FOR_EACH (row, hmap_node, &mt->table->rows) {
1830                 ovsdb_monitor_change_cb(NULL, row, NULL, &aux);
1831             }
1832         }
1833     }
1834     json = ovsdb_monitor_compose_table_update(dbmon, true);
1835     return json ? json : json_object_create();
1836 }
1837
1838 static void
1839 ovsdb_jsonrpc_monitor_destroy(struct ovsdb_jsonrpc_monitor *m)
1840 {
1841     json_destroy(m->monitor_id);
1842     hmap_remove(&m->session->monitors, &m->node);
1843     ovsdb_monitor_destroy(m->dbmon);
1844     free(m);
1845 }
1846
1847 static void
1848 ovsdb_monitor_destroy(struct ovsdb_monitor *dbmon)
1849 {
1850     struct shash_node *node;
1851
1852     list_remove(&dbmon->replica.node);
1853
1854     SHASH_FOR_EACH (node, &dbmon->tables) {
1855         struct ovsdb_monitor_table *mt = node->data;
1856         struct ovsdb_monitor_row *row, *next;
1857
1858         HMAP_FOR_EACH_SAFE (row, next, hmap_node, &mt->changes) {
1859             hmap_remove(&mt->changes, &row->hmap_node);
1860             ovsdb_monitor_row_destroy(mt, row);
1861         }
1862         hmap_destroy(&mt->changes);
1863
1864         free(mt->columns);
1865         free(mt);
1866     }
1867     shash_destroy(&dbmon->tables);
1868     free(dbmon);
1869 }
1870
1871 static void
1872 ovsdb_monitor_destroy_callback(struct ovsdb_replica *replica)
1873 {
1874     struct ovsdb_monitor *dbmon = ovsdb_monitor_cast(replica);
1875     struct ovsdb_jsonrpc_monitor *m = dbmon->jsonrpc_monitor;
1876
1877     ovsdb_jsonrpc_monitor_destroy(m);
1878 }
1879
1880 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class = {
1881     ovsdb_monitor_commit,
1882     ovsdb_monitor_destroy_callback,
1883 };