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