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 /* JSON-RPC database table monitors. */
1030
1031 enum ovsdb_jsonrpc_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_jsonrpc_monitor_column {
1040     const struct ovsdb_column *column;
1041     enum ovsdb_jsonrpc_monitor_selection select;
1042 };
1043
1044 /* A row that has changed in a monitored table. */
1045 struct ovsdb_jsonrpc_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_jsonrpc_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_jsonrpc_monitor_selection select;
1059
1060     /* Columns being monitored. */
1061     struct ovsdb_jsonrpc_monitor_column *columns;
1062     size_t n_columns;
1063
1064     /* Contains 'struct ovsdb_jsonrpc_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 /* A collection of tables being monitored. */
1070 struct ovsdb_jsonrpc_monitor {
1071     struct ovsdb_replica replica;
1072     struct ovsdb_jsonrpc_session *session;
1073     struct ovsdb *db;
1074     struct hmap_node node;      /* In ovsdb_jsonrpc_session's "monitors". */
1075
1076     struct json *monitor_id;
1077     struct shash tables;     /* Holds "struct ovsdb_jsonrpc_monitor_table"s. */
1078 };
1079
1080 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class;
1081
1082 struct ovsdb_jsonrpc_monitor *ovsdb_jsonrpc_monitor_find(
1083     struct ovsdb_jsonrpc_session *, const struct json *monitor_id);
1084 static void ovsdb_jsonrpc_monitor_destroy(struct ovsdb_replica *);
1085 static struct json *ovsdb_jsonrpc_monitor_get_initial(
1086     const struct ovsdb_jsonrpc_monitor *);
1087
1088 static bool
1089 parse_bool(struct ovsdb_parser *parser, const char *name, bool default_value)
1090 {
1091     const struct json *json;
1092
1093     json = ovsdb_parser_member(parser, name, OP_BOOLEAN | OP_OPTIONAL);
1094     return json ? json_boolean(json) : default_value;
1095 }
1096
1097 struct ovsdb_jsonrpc_monitor *
1098 ovsdb_jsonrpc_monitor_find(struct ovsdb_jsonrpc_session *s,
1099                            const struct json *monitor_id)
1100 {
1101     struct ovsdb_jsonrpc_monitor *m;
1102
1103     HMAP_FOR_EACH_WITH_HASH (m, node, json_hash(monitor_id, 0), &s->monitors) {
1104         if (json_equal(m->monitor_id, monitor_id)) {
1105             return m;
1106         }
1107     }
1108
1109     return NULL;
1110 }
1111
1112 static void
1113 ovsdb_jsonrpc_add_monitor_column(struct ovsdb_jsonrpc_monitor_table *mt,
1114                                  const struct ovsdb_column *column,
1115                                  enum ovsdb_jsonrpc_monitor_selection select,
1116                                  size_t *allocated_columns)
1117 {
1118     struct ovsdb_jsonrpc_monitor_column *c;
1119
1120     if (mt->n_columns >= *allocated_columns) {
1121         mt->columns = x2nrealloc(mt->columns, allocated_columns,
1122                                  sizeof *mt->columns);
1123     }
1124
1125     c = &mt->columns[mt->n_columns++];
1126     c->column = column;
1127     c->select = select;
1128 }
1129
1130 static int
1131 compare_ovsdb_jsonrpc_monitor_column(const void *a_, const void *b_)
1132 {
1133     const struct ovsdb_jsonrpc_monitor_column *a = a_;
1134     const struct ovsdb_jsonrpc_monitor_column *b = b_;
1135
1136     return a->column < b->column ? -1 : a->column > b->column;
1137 }
1138
1139 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
1140 ovsdb_jsonrpc_parse_monitor_request(struct ovsdb_jsonrpc_monitor_table *mt,
1141                                     const struct json *monitor_request,
1142                                     size_t *allocated_columns)
1143 {
1144     const struct ovsdb_table_schema *ts = mt->table->schema;
1145     enum ovsdb_jsonrpc_monitor_selection select;
1146     const struct json *columns, *select_json;
1147     struct ovsdb_parser parser;
1148     struct ovsdb_error *error;
1149
1150     ovsdb_parser_init(&parser, monitor_request, "table %s", ts->name);
1151     columns = ovsdb_parser_member(&parser, "columns", OP_ARRAY | OP_OPTIONAL);
1152     select_json = ovsdb_parser_member(&parser, "select",
1153                                       OP_OBJECT | OP_OPTIONAL);
1154     error = ovsdb_parser_finish(&parser);
1155     if (error) {
1156         return error;
1157     }
1158
1159     if (select_json) {
1160         select = 0;
1161         ovsdb_parser_init(&parser, select_json, "table %s select", ts->name);
1162         if (parse_bool(&parser, "initial", true)) {
1163             select |= OJMS_INITIAL;
1164         }
1165         if (parse_bool(&parser, "insert", true)) {
1166             select |= OJMS_INSERT;
1167         }
1168         if (parse_bool(&parser, "delete", true)) {
1169             select |= OJMS_DELETE;
1170         }
1171         if (parse_bool(&parser, "modify", true)) {
1172             select |= OJMS_MODIFY;
1173         }
1174         error = ovsdb_parser_finish(&parser);
1175         if (error) {
1176             return error;
1177         }
1178     } else {
1179         select = OJMS_INITIAL | OJMS_INSERT | OJMS_DELETE | OJMS_MODIFY;
1180     }
1181     mt->select |= select;
1182
1183     if (columns) {
1184         size_t i;
1185
1186         if (columns->type != JSON_ARRAY) {
1187             return ovsdb_syntax_error(columns, NULL,
1188                                       "array of column names expected");
1189         }
1190
1191         for (i = 0; i < columns->u.array.n; i++) {
1192             const struct ovsdb_column *column;
1193             const char *s;
1194
1195             if (columns->u.array.elems[i]->type != JSON_STRING) {
1196                 return ovsdb_syntax_error(columns, NULL,
1197                                           "array of column names expected");
1198             }
1199
1200             s = columns->u.array.elems[i]->u.string;
1201             column = shash_find_data(&mt->table->schema->columns, s);
1202             if (!column) {
1203                 return ovsdb_syntax_error(columns, NULL, "%s is not a valid "
1204                                           "column name", s);
1205             }
1206             ovsdb_jsonrpc_add_monitor_column(mt, column, select,
1207                                              allocated_columns);
1208         }
1209     } else {
1210         struct shash_node *node;
1211
1212         SHASH_FOR_EACH (node, &ts->columns) {
1213             const struct ovsdb_column *column = node->data;
1214             if (column->index != OVSDB_COL_UUID) {
1215                 ovsdb_jsonrpc_add_monitor_column(mt, column, select,
1216                                                  allocated_columns);
1217             }
1218         }
1219     }
1220
1221     return NULL;
1222 }
1223
1224 static struct jsonrpc_msg *
1225 ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
1226                              struct json *params,
1227                              const struct json *request_id)
1228 {
1229     struct ovsdb_jsonrpc_monitor *m = NULL;
1230     struct json *monitor_id, *monitor_requests;
1231     struct ovsdb_error *error = NULL;
1232     struct shash_node *node;
1233     struct json *json;
1234
1235     if (json_array(params)->n != 3) {
1236         error = ovsdb_syntax_error(params, NULL, "invalid parameters");
1237         goto error;
1238     }
1239     monitor_id = params->u.array.elems[1];
1240     monitor_requests = params->u.array.elems[2];
1241     if (monitor_requests->type != JSON_OBJECT) {
1242         error = ovsdb_syntax_error(monitor_requests, NULL,
1243                                    "monitor-requests must be object");
1244         goto error;
1245     }
1246
1247     if (ovsdb_jsonrpc_monitor_find(s, monitor_id)) {
1248         error = ovsdb_syntax_error(monitor_id, NULL, "duplicate monitor ID");
1249         goto error;
1250     }
1251
1252     m = xzalloc(sizeof *m);
1253     ovsdb_replica_init(&m->replica, &ovsdb_jsonrpc_replica_class);
1254     ovsdb_add_replica(db, &m->replica);
1255     m->session = s;
1256     m->db = db;
1257     hmap_insert(&s->monitors, &m->node, json_hash(monitor_id, 0));
1258     m->monitor_id = json_clone(monitor_id);
1259     shash_init(&m->tables);
1260
1261     SHASH_FOR_EACH (node, json_object(monitor_requests)) {
1262         const struct ovsdb_table *table;
1263         struct ovsdb_jsonrpc_monitor_table *mt;
1264         size_t allocated_columns;
1265         const struct json *mr_value;
1266         size_t i;
1267
1268         table = ovsdb_get_table(m->db, node->name);
1269         if (!table) {
1270             error = ovsdb_syntax_error(NULL, NULL,
1271                                        "no table named %s", node->name);
1272             goto error;
1273         }
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         /* Parse columns. */
1281         mr_value = node->data;
1282         allocated_columns = 0;
1283         if (mr_value->type == JSON_ARRAY) {
1284             const struct json_array *array = &mr_value->u.array;
1285
1286             for (i = 0; i < array->n; i++) {
1287                 error = ovsdb_jsonrpc_parse_monitor_request(
1288                     mt, array->elems[i], &allocated_columns);
1289                 if (error) {
1290                     goto error;
1291                 }
1292             }
1293         } else {
1294             error = ovsdb_jsonrpc_parse_monitor_request(
1295                 mt, mr_value, &allocated_columns);
1296             if (error) {
1297                 goto error;
1298             }
1299         }
1300
1301         /* Check for duplicate columns. */
1302         qsort(mt->columns, mt->n_columns, sizeof *mt->columns,
1303               compare_ovsdb_jsonrpc_monitor_column);
1304         for (i = 1; i < mt->n_columns; i++) {
1305             if (mt->columns[i].column == mt->columns[i - 1].column) {
1306                 error = ovsdb_syntax_error(mr_value, NULL, "column %s "
1307                                            "mentioned more than once",
1308                                            mt->columns[i].column->name);
1309                 goto error;
1310             }
1311         }
1312     }
1313
1314     return jsonrpc_create_reply(ovsdb_jsonrpc_monitor_get_initial(m),
1315                                 request_id);
1316
1317 error:
1318     if (m) {
1319         ovsdb_remove_replica(m->db, &m->replica);
1320     }
1321
1322     json = ovsdb_error_to_json(error);
1323     ovsdb_error_destroy(error);
1324     return jsonrpc_create_error(json, request_id);
1325 }
1326
1327 static struct jsonrpc_msg *
1328 ovsdb_jsonrpc_monitor_cancel(struct ovsdb_jsonrpc_session *s,
1329                              struct json_array *params,
1330                              const struct json *request_id)
1331 {
1332     if (params->n != 1) {
1333         return jsonrpc_create_error(json_string_create("invalid parameters"),
1334                                     request_id);
1335     } else {
1336         struct ovsdb_jsonrpc_monitor *m;
1337
1338         m = ovsdb_jsonrpc_monitor_find(s, params->elems[0]);
1339         if (!m) {
1340             return jsonrpc_create_error(json_string_create("unknown monitor"),
1341                                         request_id);
1342         } else {
1343             ovsdb_remove_replica(m->db, &m->replica);
1344             return jsonrpc_create_reply(json_object_create(), request_id);
1345         }
1346     }
1347 }
1348
1349 static void
1350 ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *s)
1351 {
1352     struct ovsdb_jsonrpc_monitor *m, *next;
1353
1354     HMAP_FOR_EACH_SAFE (m, next, node, &s->monitors) {
1355         ovsdb_remove_replica(m->db, &m->replica);
1356     }
1357 }
1358
1359 static struct ovsdb_jsonrpc_monitor *
1360 ovsdb_jsonrpc_monitor_cast(struct ovsdb_replica *replica)
1361 {
1362     ovs_assert(replica->class == &ovsdb_jsonrpc_replica_class);
1363     return CONTAINER_OF(replica, struct ovsdb_jsonrpc_monitor, replica);
1364 }
1365
1366 struct ovsdb_jsonrpc_monitor_aux {
1367     const struct ovsdb_jsonrpc_monitor *monitor;
1368     struct ovsdb_jsonrpc_monitor_table *mt;
1369 };
1370
1371 /* Finds and returns the ovsdb_jsonrpc_monitor_row in 'mt->changes' for the
1372  * given 'uuid', or NULL if there is no such row. */
1373 static struct ovsdb_jsonrpc_monitor_row *
1374 ovsdb_jsonrpc_monitor_row_find(const struct ovsdb_jsonrpc_monitor_table *mt,
1375                                const struct uuid *uuid)
1376 {
1377     struct ovsdb_jsonrpc_monitor_row *row;
1378
1379     HMAP_FOR_EACH_WITH_HASH (row, hmap_node, uuid_hash(uuid), &mt->changes) {
1380         if (uuid_equals(uuid, &row->uuid)) {
1381             return row;
1382         }
1383     }
1384     return NULL;
1385 }
1386
1387 /* Allocates an array of 'mt->n_columns' ovsdb_datums and initializes them as
1388  * copies of the data in 'row' drawn from the columns represented by
1389  * mt->columns[].  Returns the array.
1390  *
1391  * If 'row' is NULL, returns NULL. */
1392 static struct ovsdb_datum *
1393 clone_monitor_row_data(const struct ovsdb_jsonrpc_monitor_table *mt,
1394                        const struct ovsdb_row *row)
1395 {
1396     struct ovsdb_datum *data;
1397     size_t i;
1398
1399     if (!row) {
1400         return NULL;
1401     }
1402
1403     data = xmalloc(mt->n_columns * sizeof *data);
1404     for (i = 0; i < mt->n_columns; i++) {
1405         const struct ovsdb_column *c = mt->columns[i].column;
1406         const struct ovsdb_datum *src = &row->fields[c->index];
1407         struct ovsdb_datum *dst = &data[i];
1408         const struct ovsdb_type *type = &c->type;
1409
1410         ovsdb_datum_clone(dst, src, type);
1411     }
1412     return data;
1413 }
1414
1415 /* Replaces the mt->n_columns ovsdb_datums in row[] by copies of the data from
1416  * in 'row' drawn from the columns represented by mt->columns[]. */
1417 static void
1418 update_monitor_row_data(const struct ovsdb_jsonrpc_monitor_table *mt,
1419                         const struct ovsdb_row *row,
1420                         struct ovsdb_datum *data)
1421 {
1422     size_t i;
1423
1424     for (i = 0; i < mt->n_columns; i++) {
1425         const struct ovsdb_column *c = mt->columns[i].column;
1426         const struct ovsdb_datum *src = &row->fields[c->index];
1427         struct ovsdb_datum *dst = &data[i];
1428         const struct ovsdb_type *type = &c->type;
1429
1430         if (!ovsdb_datum_equals(src, dst, type)) {
1431             ovsdb_datum_destroy(dst, type);
1432             ovsdb_datum_clone(dst, src, type);
1433         }
1434     }
1435 }
1436
1437 /* Frees all of the mt->n_columns ovsdb_datums in data[], using the types taken
1438  * from mt->columns[], plus 'data' itself. */
1439 static void
1440 free_monitor_row_data(const struct ovsdb_jsonrpc_monitor_table *mt,
1441                       struct ovsdb_datum *data)
1442 {
1443     if (data) {
1444         size_t i;
1445
1446         for (i = 0; i < mt->n_columns; i++) {
1447             const struct ovsdb_column *c = mt->columns[i].column;
1448
1449             ovsdb_datum_destroy(&data[i], &c->type);
1450         }
1451         free(data);
1452     }
1453 }
1454
1455 /* Frees 'row', which must have been created from 'mt'. */
1456 static void
1457 ovsdb_jsonrpc_monitor_row_destroy(const struct ovsdb_jsonrpc_monitor_table *mt,
1458                                   struct ovsdb_jsonrpc_monitor_row *row)
1459 {
1460     if (row) {
1461         free_monitor_row_data(mt, row->old);
1462         free_monitor_row_data(mt, row->new);
1463         free(row);
1464     }
1465 }
1466
1467 static bool
1468 ovsdb_jsonrpc_monitor_change_cb(const struct ovsdb_row *old,
1469                                 const struct ovsdb_row *new,
1470                                 const unsigned long int *changed OVS_UNUSED,
1471                                 void *aux_)
1472 {
1473     struct ovsdb_jsonrpc_monitor_aux *aux = aux_;
1474     const struct ovsdb_jsonrpc_monitor *m = aux->monitor;
1475     struct ovsdb_table *table = new ? new->table : old->table;
1476     const struct uuid *uuid = ovsdb_row_get_uuid(new ? new : old);
1477     struct ovsdb_jsonrpc_monitor_row *change;
1478     struct ovsdb_jsonrpc_monitor_table *mt;
1479
1480     if (!aux->mt || table != aux->mt->table) {
1481         aux->mt = shash_find_data(&m->tables, table->schema->name);
1482         if (!aux->mt) {
1483             /* We don't care about rows in this table at all.  Tell the caller
1484              * to skip it.  */
1485             return false;
1486         }
1487     }
1488     mt = aux->mt;
1489
1490     change = ovsdb_jsonrpc_monitor_row_find(mt, uuid);
1491     if (!change) {
1492         change = xmalloc(sizeof *change);
1493         hmap_insert(&mt->changes, &change->hmap_node, uuid_hash(uuid));
1494         change->uuid = *uuid;
1495         change->old = clone_monitor_row_data(mt, old);
1496         change->new = clone_monitor_row_data(mt, new);
1497     } else {
1498         if (new) {
1499             update_monitor_row_data(mt, new, change->new);
1500         } else {
1501             free_monitor_row_data(mt, change->new);
1502             change->new = NULL;
1503
1504             if (!change->old) {
1505                 /* This row was added then deleted.  Forget about it. */
1506                 hmap_remove(&mt->changes, &change->hmap_node);
1507                 free(change);
1508             }
1509         }
1510     }
1511     return true;
1512 }
1513
1514 /* Returns JSON for a <row-update> (as described in RFC 7047) for 'row' within
1515  * 'mt', or NULL if no row update should be sent.
1516  *
1517  * The caller should specify 'initial' as true if the returned JSON is going to
1518  * be used as part of the initial reply to a "monitor" request, false if it is
1519  * going to be used as part of an "update" notification.
1520  *
1521  * 'changed' must be a scratch buffer for internal use that is at least
1522  * bitmap_n_bytes(mt->n_columns) bytes long. */
1523 static struct json *
1524 ovsdb_jsonrpc_monitor_compose_row_update(
1525     const struct ovsdb_jsonrpc_monitor_table *mt,
1526     const struct ovsdb_jsonrpc_monitor_row *row,
1527     bool initial, unsigned long int *changed)
1528 {
1529     enum ovsdb_jsonrpc_monitor_selection type;
1530     struct json *old_json, *new_json;
1531     struct json *row_json;
1532     size_t i;
1533
1534     type = (initial ? OJMS_INITIAL
1535             : !row->old ? OJMS_INSERT
1536             : !row->new ? OJMS_DELETE
1537             : OJMS_MODIFY);
1538     if (!(mt->select & type)) {
1539         return NULL;
1540     }
1541
1542     if (type == OJMS_MODIFY) {
1543         size_t n_changes;
1544
1545         n_changes = 0;
1546         memset(changed, 0, bitmap_n_bytes(mt->n_columns));
1547         for (i = 0; i < mt->n_columns; i++) {
1548             const struct ovsdb_column *c = mt->columns[i].column;
1549             if (!ovsdb_datum_equals(&row->old[i], &row->new[i], &c->type)) {
1550                 bitmap_set1(changed, i);
1551                 n_changes++;
1552             }
1553         }
1554         if (!n_changes) {
1555             /* No actual changes: presumably a row changed and then
1556              * changed back later. */
1557             return NULL;
1558         }
1559     }
1560
1561     row_json = json_object_create();
1562     old_json = new_json = NULL;
1563     if (type & (OJMS_DELETE | OJMS_MODIFY)) {
1564         old_json = json_object_create();
1565         json_object_put(row_json, "old", old_json);
1566     }
1567     if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
1568         new_json = json_object_create();
1569         json_object_put(row_json, "new", new_json);
1570     }
1571     for (i = 0; i < mt->n_columns; i++) {
1572         const struct ovsdb_jsonrpc_monitor_column *c = &mt->columns[i];
1573
1574         if (!(type & c->select)) {
1575             /* We don't care about this type of change for this
1576              * particular column (but we will care about it for some
1577              * other column). */
1578             continue;
1579         }
1580
1581         if ((type == OJMS_MODIFY && bitmap_is_set(changed, i))
1582             || type == OJMS_DELETE) {
1583             json_object_put(old_json, c->column->name,
1584                             ovsdb_datum_to_json(&row->old[i],
1585                                                 &c->column->type));
1586         }
1587         if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
1588             json_object_put(new_json, c->column->name,
1589                             ovsdb_datum_to_json(&row->new[i],
1590                                                 &c->column->type));
1591         }
1592     }
1593
1594     return row_json;
1595 }
1596
1597 /* Constructs and returns JSON for a <table-updates> object (as described in
1598  * RFC 7047) for all the outstanding changes within 'monitor', and deletes all
1599  * the outstanding changes from 'monitor'.  Returns NULL if no update needs to
1600  * be sent.
1601  *
1602  * The caller should specify 'initial' as true if the returned JSON is going to
1603  * be used as part of the initial reply to a "monitor" request, false if it is
1604  * going to be used as part of an "update" notification. */
1605 static struct json *
1606 ovsdb_jsonrpc_monitor_compose_table_update(
1607     const struct ovsdb_jsonrpc_monitor *monitor, bool initial)
1608 {
1609     struct shash_node *node;
1610     unsigned long int *changed;
1611     struct json *json;
1612     size_t max_columns;
1613
1614     max_columns = 0;
1615     SHASH_FOR_EACH (node, &monitor->tables) {
1616         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
1617
1618         max_columns = MAX(max_columns, mt->n_columns);
1619     }
1620     changed = xmalloc(bitmap_n_bytes(max_columns));
1621
1622     json = NULL;
1623     SHASH_FOR_EACH (node, &monitor->tables) {
1624         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
1625         struct ovsdb_jsonrpc_monitor_row *row, *next;
1626         struct json *table_json = NULL;
1627
1628         HMAP_FOR_EACH_SAFE (row, next, hmap_node, &mt->changes) {
1629             struct json *row_json;
1630
1631             row_json = ovsdb_jsonrpc_monitor_compose_row_update(
1632                 mt, row, initial, changed);
1633             if (row_json) {
1634                 char uuid[UUID_LEN + 1];
1635
1636                 /* Create JSON object for transaction overall. */
1637                 if (!json) {
1638                     json = json_object_create();
1639                 }
1640
1641                 /* Create JSON object for transaction on this table. */
1642                 if (!table_json) {
1643                     table_json = json_object_create();
1644                     json_object_put(json, mt->table->schema->name, table_json);
1645                 }
1646
1647                 /* Add JSON row to JSON table. */
1648                 snprintf(uuid, sizeof uuid, UUID_FMT, UUID_ARGS(&row->uuid));
1649                 json_object_put(table_json, uuid, row_json);
1650             }
1651
1652             hmap_remove(&mt->changes, &row->hmap_node);
1653             ovsdb_jsonrpc_monitor_row_destroy(mt, row);
1654         }
1655     }
1656
1657     free(changed);
1658
1659     return json;
1660 }
1661
1662 static bool
1663 ovsdb_jsonrpc_monitor_needs_flush(struct ovsdb_jsonrpc_session *s)
1664 {
1665     struct ovsdb_jsonrpc_monitor *m;
1666
1667     HMAP_FOR_EACH (m, node, &s->monitors) {
1668         struct shash_node *node;
1669
1670         SHASH_FOR_EACH (node, &m->tables) {
1671             struct ovsdb_jsonrpc_monitor_table *mt = node->data;
1672
1673             if (!hmap_is_empty(&mt->changes)) {
1674                 return true;
1675             }
1676         }
1677     }
1678
1679     return false;
1680 }
1681
1682 static void
1683 ovsdb_jsonrpc_monitor_flush_all(struct ovsdb_jsonrpc_session *s)
1684 {
1685     struct ovsdb_jsonrpc_monitor *m;
1686
1687     HMAP_FOR_EACH (m, node, &s->monitors) {
1688         struct json *json;
1689
1690         json = ovsdb_jsonrpc_monitor_compose_table_update(m, false);
1691         if (json) {
1692             struct jsonrpc_msg *msg;
1693             struct json *params;
1694
1695             params = json_array_create_2(json_clone(m->monitor_id), json);
1696             msg = jsonrpc_create_notify("update", params);
1697             jsonrpc_session_send(s->js, msg);
1698         }
1699     }
1700 }
1701
1702 static void
1703 ovsdb_jsonrpc_monitor_init_aux(struct ovsdb_jsonrpc_monitor_aux *aux,
1704                                const struct ovsdb_jsonrpc_monitor *m)
1705 {
1706     aux->monitor = m;
1707     aux->mt = NULL;
1708 }
1709
1710 static struct ovsdb_error *
1711 ovsdb_jsonrpc_monitor_commit(struct ovsdb_replica *replica,
1712                              const struct ovsdb_txn *txn,
1713                              bool durable OVS_UNUSED)
1714 {
1715     struct ovsdb_jsonrpc_monitor *m = ovsdb_jsonrpc_monitor_cast(replica);
1716     struct ovsdb_jsonrpc_monitor_aux aux;
1717
1718     ovsdb_jsonrpc_monitor_init_aux(&aux, m);
1719     ovsdb_txn_for_each_change(txn, ovsdb_jsonrpc_monitor_change_cb, &aux);
1720
1721     return NULL;
1722 }
1723
1724 static struct json *
1725 ovsdb_jsonrpc_monitor_get_initial(const struct ovsdb_jsonrpc_monitor *m)
1726 {
1727     struct ovsdb_jsonrpc_monitor_aux aux;
1728     struct shash_node *node;
1729     struct json *json;
1730
1731     ovsdb_jsonrpc_monitor_init_aux(&aux, m);
1732     SHASH_FOR_EACH (node, &m->tables) {
1733         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
1734
1735         if (mt->select & OJMS_INITIAL) {
1736             struct ovsdb_row *row;
1737
1738             HMAP_FOR_EACH (row, hmap_node, &mt->table->rows) {
1739                 ovsdb_jsonrpc_monitor_change_cb(NULL, row, NULL, &aux);
1740             }
1741         }
1742     }
1743     json = ovsdb_jsonrpc_monitor_compose_table_update(m, true);
1744     return json ? json : json_object_create();
1745 }
1746
1747 static void
1748 ovsdb_jsonrpc_monitor_destroy(struct ovsdb_replica *replica)
1749 {
1750     struct ovsdb_jsonrpc_monitor *m = ovsdb_jsonrpc_monitor_cast(replica);
1751     struct shash_node *node;
1752
1753     json_destroy(m->monitor_id);
1754     SHASH_FOR_EACH (node, &m->tables) {
1755         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
1756         struct ovsdb_jsonrpc_monitor_row *row, *next;
1757
1758         HMAP_FOR_EACH_SAFE (row, next, hmap_node, &mt->changes) {
1759             hmap_remove(&mt->changes, &row->hmap_node);
1760             ovsdb_jsonrpc_monitor_row_destroy(mt, row);
1761         }
1762         hmap_destroy(&mt->changes);
1763
1764         free(mt->columns);
1765         free(mt);
1766     }
1767     shash_destroy(&m->tables);
1768     hmap_remove(&m->session->monitors, &m->node);
1769     free(m);
1770 }
1771
1772 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class = {
1773     ovsdb_jsonrpc_monitor_commit,
1774     ovsdb_jsonrpc_monitor_destroy
1775 };