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 "openvswitch/dynamic-string.h"
  25 #include "monitor.h"
  26 #include "json.h"
  27 #include "jsonrpc.h"
  28 #include "ovsdb-error.h"
  29 #include "ovsdb-parser.h"
  30 #include "ovsdb.h"
  31 #include "condition.h"
  32 #include "poll-loop.h"
  33 #include "reconnect.h"
  34 #include "row.h"
  35 #include "server.h"
  36 #include "simap.h"
  37 #include "stream.h"
  38 #include "table.h"
  39 #include "timeval.h"
  40 #include "transaction.h"
  41 #include "trigger.h"
  42 #include "openvswitch/vlog.h"
  43
  44 VLOG_DEFINE_THIS_MODULE(ovsdb_jsonrpc_server);
  45
  46 struct ovsdb_jsonrpc_remote;
  47 struct ovsdb_jsonrpc_session;
  48
  49 /* Set false to defeature monitor2, causing jsonrpc to respond to monitor2
  50  * method with an error.  */
  51 static bool monitor2_enable__ = true;
  52
  53 /* Message rate-limiting. */
  54 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
  55
  56 /* Sessions. */
  57 static struct ovsdb_jsonrpc_session *ovsdb_jsonrpc_session_create(
  58     struct ovsdb_jsonrpc_remote *, struct jsonrpc_session *);
  59 static void ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_remote *);
  60 static void ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_remote *);
  61 static void ovsdb_jsonrpc_session_get_memory_usage_all(
  62     const struct ovsdb_jsonrpc_remote *, struct simap *usage);
  63 static void ovsdb_jsonrpc_session_close_all(struct ovsdb_jsonrpc_remote *);
  64 static void ovsdb_jsonrpc_session_reconnect_all(struct ovsdb_jsonrpc_remote *);
  65 static void ovsdb_jsonrpc_session_set_all_options(
  66     struct ovsdb_jsonrpc_remote *, const struct ovsdb_jsonrpc_options *);
  67 static bool ovsdb_jsonrpc_active_session_get_status(
  68     const struct ovsdb_jsonrpc_remote *,
  69     struct ovsdb_jsonrpc_remote_status *);
  70 static void ovsdb_jsonrpc_session_get_status(
  71     const struct ovsdb_jsonrpc_session *,
  72     struct ovsdb_jsonrpc_remote_status *);
  73 static void ovsdb_jsonrpc_session_unlock_all(struct ovsdb_jsonrpc_session *);
  74 static void ovsdb_jsonrpc_session_unlock__(struct ovsdb_lock_waiter *);
  75 static void ovsdb_jsonrpc_session_send(struct ovsdb_jsonrpc_session *,
  76                                        struct jsonrpc_msg *);
  77
  78 /* Triggers. */
  79 static void ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *,
  80                                          struct ovsdb *,
  81                                          struct json *id, struct json *params);
  82 static struct ovsdb_jsonrpc_trigger *ovsdb_jsonrpc_trigger_find(
  83     struct ovsdb_jsonrpc_session *, const struct json *id, size_t hash);
  84 static void ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *);
  85 static void ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *);
  86 static void ovsdb_jsonrpc_trigger_complete_done(
  87     struct ovsdb_jsonrpc_session *);
  88
  89 /* Monitors. */
  90 static struct jsonrpc_msg *ovsdb_jsonrpc_monitor_create(
  91     struct ovsdb_jsonrpc_session *, struct ovsdb *, struct json *params,
  92     enum ovsdb_monitor_version, const struct json *request_id);
  93 static struct jsonrpc_msg *ovsdb_jsonrpc_monitor_cancel(
  94     struct ovsdb_jsonrpc_session *,
  95     struct json_array *params,
  96     const struct json *request_id);
  97 static void ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *);
  98 static void ovsdb_jsonrpc_monitor_flush_all(struct ovsdb_jsonrpc_session *);
  99 static bool ovsdb_jsonrpc_monitor_needs_flush(struct ovsdb_jsonrpc_session *);
 100 static struct json *ovsdb_jsonrpc_monitor_compose_update(
 101     struct ovsdb_jsonrpc_monitor *monitor, bool initial);
 102
 103 \f
 104 /* JSON-RPC database server. */
 105
 106 struct ovsdb_jsonrpc_server {
 107     struct ovsdb_server up;
 108     unsigned int n_sessions;
 109     struct shash remotes;      /* Contains "struct ovsdb_jsonrpc_remote *"s. */
 110 };
 111
 112 /* A configured remote.  This is either a passive stream listener plus a list
 113  * of the currently connected sessions, or a list of exactly one active
 114  * session. */
 115 struct ovsdb_jsonrpc_remote {
 116     struct ovsdb_jsonrpc_server *server;
 117     struct pstream *listener;   /* Listener, if passive. */
 118     struct ovs_list sessions;   /* List of "struct ovsdb_jsonrpc_session"s. */
 119     uint8_t dscp;
 120 };
 121
 122 static struct ovsdb_jsonrpc_remote *ovsdb_jsonrpc_server_add_remote(
 123     struct ovsdb_jsonrpc_server *, const char *name,
 124     const struct ovsdb_jsonrpc_options *options
 125 );
 126 static void ovsdb_jsonrpc_server_del_remote(struct shash_node *);
 127
 128 /* Creates and returns a new server to provide JSON-RPC access to an OVSDB.
 129  *
 130  * The caller must call ovsdb_jsonrpc_server_add_db() for each database to
 131  * which 'server' should provide access. */
 132 struct ovsdb_jsonrpc_server *
 133 ovsdb_jsonrpc_server_create(void)
 134 {
 135     struct ovsdb_jsonrpc_server *server = xzalloc(sizeof *server);
 136     ovsdb_server_init(&server->up);
 137     shash_init(&server->remotes);
 138     return server;
 139 }
 140
 141 /* Adds 'db' to the set of databases served out by 'svr'.  Returns true if
 142  * successful, false if 'db''s name is the same as some database already in
 143  * 'server'. */
 144 bool
 145 ovsdb_jsonrpc_server_add_db(struct ovsdb_jsonrpc_server *svr, struct ovsdb *db)
 146 {
 147     /* The OVSDB protocol doesn't have a way to notify a client that a
 148      * database has been added.  If some client tried to use the database
 149      * that we're adding and failed, then forcing it to reconnect seems like
 150      * a reasonable way to make it try again.
 151      *
 152      * If this is too big of a hammer in practice, we could be more selective,
 153      * e.g. disconnect only connections that actually tried to use a database
 154      * with 'db''s name. */
 155     ovsdb_jsonrpc_server_reconnect(svr);
 156
 157     return ovsdb_server_add_db(&svr->up, db);
 158 }
 159
 160 /* Removes 'db' from the set of databases served out by 'svr'.  Returns
 161  * true if successful, false if there is no database associated with 'db'. */
 162 bool
 163 ovsdb_jsonrpc_server_remove_db(struct ovsdb_jsonrpc_server *svr,
 164                                struct ovsdb *db)
 165 {
 166     /* There might be pointers to 'db' from 'svr', such as monitors or
 167      * outstanding transactions.  Disconnect all JSON-RPC connections to avoid
 168      * accesses to freed memory.
 169      *
 170      * If this is too big of a hammer in practice, we could be more selective,
 171      * e.g. disconnect only connections that actually reference 'db'. */
 172     ovsdb_jsonrpc_server_reconnect(svr);
 173
 174     return ovsdb_server_remove_db(&svr->up, db);
 175 }
 176
 177 void
 178 ovsdb_jsonrpc_server_destroy(struct ovsdb_jsonrpc_server *svr)
 179 {
 180     struct shash_node *node, *next;
 181
 182     SHASH_FOR_EACH_SAFE (node, next, &svr->remotes) {
 183         ovsdb_jsonrpc_server_del_remote(node);
 184     }
 185     shash_destroy(&svr->remotes);
 186     ovsdb_server_destroy(&svr->up);
 187     free(svr);
 188 }
 189
 190 struct ovsdb_jsonrpc_options *
 191 ovsdb_jsonrpc_default_options(const char *target)
 192 {
 193     struct ovsdb_jsonrpc_options *options = xzalloc(sizeof *options);
 194     options->max_backoff = RECONNECT_DEFAULT_MAX_BACKOFF;
 195     options->probe_interval = (stream_or_pstream_needs_probes(target)
 196                                ? RECONNECT_DEFAULT_PROBE_INTERVAL
 197                                : 0);
 198     return options;
 199 }
 200
 201 /* Sets 'svr''s current set of remotes to the names in 'new_remotes', with
 202  * options in the struct ovsdb_jsonrpc_options supplied as the data values.
 203  *
 204  * A remote is an active or passive stream connection method, e.g. "pssl:" or
 205  * "tcp:1.2.3.4". */
 206 void
 207 ovsdb_jsonrpc_server_set_remotes(struct ovsdb_jsonrpc_server *svr,
 208                                  const struct shash *new_remotes)
 209 {
 210     struct shash_node *node, *next;
 211
 212     SHASH_FOR_EACH_SAFE (node, next, &svr->remotes) {
 213         struct ovsdb_jsonrpc_remote *remote = node->data;
 214         struct ovsdb_jsonrpc_options *options
 215             = shash_find_data(new_remotes, node->name);
 216
 217         if (!options) {
 218             VLOG_INFO("%s: remote deconfigured", node->name);
 219             ovsdb_jsonrpc_server_del_remote(node);
 220         } else if (options->dscp != remote->dscp) {
 221             ovsdb_jsonrpc_server_del_remote(node);
 222          }
 223     }
 224     SHASH_FOR_EACH (node, new_remotes) {
 225         const struct ovsdb_jsonrpc_options *options = node->data;
 226         struct ovsdb_jsonrpc_remote *remote;
 227
 228         remote = shash_find_data(&svr->remotes, node->name);
 229         if (!remote) {
 230             remote = ovsdb_jsonrpc_server_add_remote(svr, node->name, options);
 231             if (!remote) {
 232                 continue;
 233             }
 234         }
 235
 236         ovsdb_jsonrpc_session_set_all_options(remote, options);
 237     }
 238 }
 239
 240 static struct ovsdb_jsonrpc_remote *
 241 ovsdb_jsonrpc_server_add_remote(struct ovsdb_jsonrpc_server *svr,
 242                                 const char *name,
 243                                 const struct ovsdb_jsonrpc_options *options)
 244 {
 245     struct ovsdb_jsonrpc_remote *remote;
 246     struct pstream *listener;
 247     int error;
 248
 249     error = jsonrpc_pstream_open(name, &listener, options->dscp);
 250     if (error && error != EAFNOSUPPORT) {
 251         VLOG_ERR_RL(&rl, "%s: listen failed: %s", name, ovs_strerror(error));
 252         return NULL;
 253     }
 254
 255     remote = xmalloc(sizeof *remote);
 256     remote->server = svr;
 257     remote->listener = listener;
 258     ovs_list_init(&remote->sessions);
 259     remote->dscp = options->dscp;
 260     shash_add(&svr->remotes, name, remote);
 261
 262     if (!listener) {
 263         ovsdb_jsonrpc_session_create(remote, jsonrpc_session_open(name, true));
 264     }
 265     return remote;
 266 }
 267
 268 static void
 269 ovsdb_jsonrpc_server_del_remote(struct shash_node *node)
 270 {
 271     struct ovsdb_jsonrpc_remote *remote = node->data;
 272
 273     ovsdb_jsonrpc_session_close_all(remote);
 274     pstream_close(remote->listener);
 275     shash_delete(&remote->server->remotes, node);
 276     free(remote);
 277 }
 278
 279 /* Stores status information for the remote named 'target', which should have
 280  * been configured on 'svr' with a call to ovsdb_jsonrpc_server_set_remotes(),
 281  * into '*status'.  On success returns true, on failure (if 'svr' doesn't have
 282  * a remote named 'target' or if that remote is an outbound remote that has no
 283  * active connections) returns false.  On failure, 'status' will be zeroed.
 284  */
 285 bool
 286 ovsdb_jsonrpc_server_get_remote_status(
 287     const struct ovsdb_jsonrpc_server *svr, const char *target,
 288     struct ovsdb_jsonrpc_remote_status *status)
 289 {
 290     const struct ovsdb_jsonrpc_remote *remote;
 291
 292     memset(status, 0, sizeof *status);
 293
 294     remote = shash_find_data(&svr->remotes, target);
 295
 296     if (!remote) {
 297         return false;
 298     }
 299
 300     if (remote->listener) {
 301         status->bound_port = pstream_get_bound_port(remote->listener);
 302         status->is_connected = !ovs_list_is_empty(&remote->sessions);
 303         status->n_connections = ovs_list_size(&remote->sessions);
 304         return true;
 305     }
 306
 307     return ovsdb_jsonrpc_active_session_get_status(remote, status);
 308 }
 309
 310 void
 311 ovsdb_jsonrpc_server_free_remote_status(
 312     struct ovsdb_jsonrpc_remote_status *status)
 313 {
 314     free(status->locks_held);
 315     free(status->locks_waiting);
 316     free(status->locks_lost);
 317 }
 318
 319 /* Forces all of the JSON-RPC sessions managed by 'svr' to disconnect and
 320  * reconnect. */
 321 void
 322 ovsdb_jsonrpc_server_reconnect(struct ovsdb_jsonrpc_server *svr)
 323 {
 324     struct shash_node *node;
 325
 326     SHASH_FOR_EACH (node, &svr->remotes) {
 327         struct ovsdb_jsonrpc_remote *remote = node->data;
 328
 329         ovsdb_jsonrpc_session_reconnect_all(remote);
 330     }
 331 }
 332
 333 void
 334 ovsdb_jsonrpc_server_run(struct ovsdb_jsonrpc_server *svr)
 335 {
 336     struct shash_node *node;
 337
 338     SHASH_FOR_EACH (node, &svr->remotes) {
 339         struct ovsdb_jsonrpc_remote *remote = node->data;
 340
 341         if (remote->listener) {
 342             struct stream *stream;
 343             int error;
 344
 345             error = pstream_accept(remote->listener, &stream);
 346             if (!error) {
 347                 struct jsonrpc_session *js;
 348                 js = jsonrpc_session_open_unreliably(jsonrpc_open(stream),
 349                                                      remote->dscp);
 350                 ovsdb_jsonrpc_session_create(remote, js);
 351             } else if (error != EAGAIN) {
 352                 VLOG_WARN_RL(&rl, "%s: accept failed: %s",
 353                              pstream_get_name(remote->listener),
 354                              ovs_strerror(error));
 355             }
 356         }
 357
 358         ovsdb_jsonrpc_session_run_all(remote);
 359     }
 360 }
 361
 362 void
 363 ovsdb_jsonrpc_server_wait(struct ovsdb_jsonrpc_server *svr)
 364 {
 365     struct shash_node *node;
 366
 367     SHASH_FOR_EACH (node, &svr->remotes) {
 368         struct ovsdb_jsonrpc_remote *remote = node->data;
 369
 370         if (remote->listener) {
 371             pstream_wait(remote->listener);
 372         }
 373
 374         ovsdb_jsonrpc_session_wait_all(remote);
 375     }
 376 }
 377
 378 /* Adds some memory usage statistics for 'svr' into 'usage', for use with
 379  * memory_report(). */
 380 void
 381 ovsdb_jsonrpc_server_get_memory_usage(const struct ovsdb_jsonrpc_server *svr,
 382                                       struct simap *usage)
 383 {
 384     struct shash_node *node;
 385
 386     simap_increase(usage, "sessions", svr->n_sessions);
 387     SHASH_FOR_EACH (node, &svr->remotes) {
 388         struct ovsdb_jsonrpc_remote *remote = node->data;
 389
 390         ovsdb_jsonrpc_session_get_memory_usage_all(remote, usage);
 391     }
 392 }
 393 \f
 394 /* JSON-RPC database server session. */
 395
 396 struct ovsdb_jsonrpc_session {
 397     struct ovs_list node;       /* Element in remote's sessions list. */
 398     struct ovsdb_session up;
 399     struct ovsdb_jsonrpc_remote *remote;
 400
 401     /* Triggers. */
 402     struct hmap triggers;       /* Hmap of "struct ovsdb_jsonrpc_trigger"s. */
 403
 404     /* Monitors. */
 405     struct hmap monitors;       /* Hmap of "struct ovsdb_jsonrpc_monitor"s. */
 406
 407     /* Network connectivity. */
 408     struct jsonrpc_session *js;  /* JSON-RPC session. */
 409     unsigned int js_seqno;       /* Last jsonrpc_session_get_seqno() value. */
 410 };
 411
 412 static void ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *);
 413 static int ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *);
 414 static void ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *);
 415 static void ovsdb_jsonrpc_session_get_memory_usage(
 416     const struct ovsdb_jsonrpc_session *, struct simap *usage);
 417 static void ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *,
 418                                              struct jsonrpc_msg *);
 419 static void ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *,
 420                                              struct jsonrpc_msg *);
 421
 422 static struct ovsdb_jsonrpc_session *
 423 ovsdb_jsonrpc_session_create(struct ovsdb_jsonrpc_remote *remote,
 424                              struct jsonrpc_session *js)
 425 {
 426     struct ovsdb_jsonrpc_session *s;
 427
 428     s = xzalloc(sizeof *s);
 429     ovsdb_session_init(&s->up, &remote->server->up);
 430     s->remote = remote;
 431     ovs_list_push_back(&remote->sessions, &s->node);
 432     hmap_init(&s->triggers);
 433     hmap_init(&s->monitors);
 434     s->js = js;
 435     s->js_seqno = jsonrpc_session_get_seqno(js);
 436
 437     remote->server->n_sessions++;
 438
 439     return s;
 440 }
 441
 442 static void
 443 ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *s)
 444 {
 445     ovsdb_jsonrpc_monitor_remove_all(s);
 446     ovsdb_jsonrpc_session_unlock_all(s);
 447     ovsdb_jsonrpc_trigger_complete_all(s);
 448
 449     hmap_destroy(&s->monitors);
 450     hmap_destroy(&s->triggers);
 451
 452     jsonrpc_session_close(s->js);
 453     ovs_list_remove(&s->node);
 454     s->remote->server->n_sessions--;
 455     ovsdb_session_destroy(&s->up);
 456     free(s);
 457 }
 458
 459 static int
 460 ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *s)
 461 {
 462     jsonrpc_session_run(s->js);
 463     if (s->js_seqno != jsonrpc_session_get_seqno(s->js)) {
 464         s->js_seqno = jsonrpc_session_get_seqno(s->js);
 465         ovsdb_jsonrpc_trigger_complete_all(s);
 466         ovsdb_jsonrpc_monitor_remove_all(s);
 467         ovsdb_jsonrpc_session_unlock_all(s);
 468     }
 469
 470     ovsdb_jsonrpc_trigger_complete_done(s);
 471
 472     if (!jsonrpc_session_get_backlog(s->js)) {
 473         struct jsonrpc_msg *msg;
 474
 475         ovsdb_jsonrpc_monitor_flush_all(s);
 476
 477         msg = jsonrpc_session_recv(s->js);
 478         if (msg) {
 479             if (msg->type == JSONRPC_REQUEST) {
 480                 ovsdb_jsonrpc_session_got_request(s, msg);
 481             } else if (msg->type == JSONRPC_NOTIFY) {
 482                 ovsdb_jsonrpc_session_got_notify(s, msg);
 483             } else {
 484                 VLOG_WARN("%s: received unexpected %s message",
 485                           jsonrpc_session_get_name(s->js),
 486                           jsonrpc_msg_type_to_string(msg->type));
 487                 jsonrpc_session_force_reconnect(s->js);
 488                 jsonrpc_msg_destroy(msg);
 489             }
 490         }
 491     }
 492     return jsonrpc_session_is_alive(s->js) ? 0 : ETIMEDOUT;
 493 }
 494
 495 static void
 496 ovsdb_jsonrpc_session_set_options(struct ovsdb_jsonrpc_session *session,
 497                                   const struct ovsdb_jsonrpc_options *options)
 498 {
 499     jsonrpc_session_set_max_backoff(session->js, options->max_backoff);
 500     jsonrpc_session_set_probe_interval(session->js, options->probe_interval);
 501     jsonrpc_session_set_dscp(session->js, options->dscp);
 502 }
 503
 504 static void
 505 ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_remote *remote)
 506 {
 507     struct ovsdb_jsonrpc_session *s, *next;
 508
 509     LIST_FOR_EACH_SAFE (s, next, node, &remote->sessions) {
 510         int error = ovsdb_jsonrpc_session_run(s);
 511         if (error) {
 512             ovsdb_jsonrpc_session_close(s);
 513         }
 514     }
 515 }
 516
 517 static void
 518 ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *s)
 519 {
 520     jsonrpc_session_wait(s->js);
 521     if (!jsonrpc_session_get_backlog(s->js)) {
 522         if (ovsdb_jsonrpc_monitor_needs_flush(s)) {
 523             poll_immediate_wake();
 524         } else {
 525             jsonrpc_session_recv_wait(s->js);
 526         }
 527     }
 528 }
 529
 530 static void
 531 ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_remote *remote)
 532 {
 533     struct ovsdb_jsonrpc_session *s;
 534
 535     LIST_FOR_EACH (s, node, &remote->sessions) {
 536         ovsdb_jsonrpc_session_wait(s);
 537     }
 538 }
 539
 540 static void
 541 ovsdb_jsonrpc_session_get_memory_usage(const struct ovsdb_jsonrpc_session *s,
 542                                        struct simap *usage)
 543 {
 544     simap_increase(usage, "triggers", hmap_count(&s->triggers));
 545     simap_increase(usage, "backlog", jsonrpc_session_get_backlog(s->js));
 546 }
 547
 548 static void
 549 ovsdb_jsonrpc_session_get_memory_usage_all(
 550     const struct ovsdb_jsonrpc_remote *remote,
 551     struct simap *usage)
 552 {
 553     struct ovsdb_jsonrpc_session *s;
 554
 555     LIST_FOR_EACH (s, node, &remote->sessions) {
 556         ovsdb_jsonrpc_session_get_memory_usage(s, usage);
 557     }
 558 }
 559
 560 static void
 561 ovsdb_jsonrpc_session_close_all(struct ovsdb_jsonrpc_remote *remote)
 562 {
 563     struct ovsdb_jsonrpc_session *s, *next;
 564
 565     LIST_FOR_EACH_SAFE (s, next, node, &remote->sessions) {
 566         ovsdb_jsonrpc_session_close(s);
 567     }
 568 }
 569
 570 /* Forces all of the JSON-RPC sessions managed by 'remote' to disconnect and
 571  * reconnect. */
 572 static void
 573 ovsdb_jsonrpc_session_reconnect_all(struct ovsdb_jsonrpc_remote *remote)
 574 {
 575     struct ovsdb_jsonrpc_session *s, *next;
 576
 577     LIST_FOR_EACH_SAFE (s, next, node, &remote->sessions) {
 578         jsonrpc_session_force_reconnect(s->js);
 579         if (!jsonrpc_session_is_alive(s->js)) {
 580             ovsdb_jsonrpc_session_close(s);
 581         }
 582     }
 583 }
 584
 585 /* Sets the options for all of the JSON-RPC sessions managed by 'remote' to
 586  * 'options'.
 587  *
 588  * (The dscp value can't be changed directly; the caller must instead close and
 589  * re-open the session.) */
 590 static void
 591 ovsdb_jsonrpc_session_set_all_options(
 592     struct ovsdb_jsonrpc_remote *remote,
 593     const struct ovsdb_jsonrpc_options *options)
 594 {
 595     struct ovsdb_jsonrpc_session *s;
 596
 597     LIST_FOR_EACH (s, node, &remote->sessions) {
 598         ovsdb_jsonrpc_session_set_options(s, options);
 599     }
 600 }
 601
 602 /* Sets the 'status' of for the 'remote' with an outgoing connection.   */
 603 static bool
 604 ovsdb_jsonrpc_active_session_get_status(
 605     const struct ovsdb_jsonrpc_remote *remote,
 606     struct ovsdb_jsonrpc_remote_status *status)
 607 {
 608     const struct ovs_list *sessions = &remote->sessions;
 609     const struct ovsdb_jsonrpc_session *s;
 610
 611     if (ovs_list_is_empty(sessions)) {
 612         return false;
 613     }
 614
 615     ovs_assert(ovs_list_is_singleton(sessions));
 616     s = CONTAINER_OF(ovs_list_front(sessions), struct ovsdb_jsonrpc_session, node);
 617     ovsdb_jsonrpc_session_get_status(s, status);
 618     status->n_connections = 1;
 619
 620     return true;
 621 }
 622
 623 static void
 624 ovsdb_jsonrpc_session_get_status(const struct ovsdb_jsonrpc_session *session,
 625                                  struct ovsdb_jsonrpc_remote_status *status)
 626 {
 627     const struct ovsdb_jsonrpc_session *s = session;
 628     const struct jsonrpc_session *js;
 629     struct ovsdb_lock_waiter *waiter;
 630     struct reconnect_stats rstats;
 631     struct ds locks_held, locks_waiting, locks_lost;
 632
 633     js = s->js;
 634
 635     status->is_connected = jsonrpc_session_is_connected(js);
 636     status->last_error = jsonrpc_session_get_status(js);
 637
 638     jsonrpc_session_get_reconnect_stats(js, &rstats);
 639     status->state = rstats.state;
 640     status->sec_since_connect = rstats.msec_since_connect == UINT_MAX
 641         ? UINT_MAX : rstats.msec_since_connect / 1000;
 642     status->sec_since_disconnect = rstats.msec_since_disconnect == UINT_MAX
 643         ? UINT_MAX : rstats.msec_since_disconnect / 1000;
 644
 645     ds_init(&locks_held);
 646     ds_init(&locks_waiting);
 647     ds_init(&locks_lost);
 648     HMAP_FOR_EACH (waiter, session_node, &s->up.waiters) {
 649         struct ds *string;
 650
 651         string = (ovsdb_lock_waiter_is_owner(waiter) ? &locks_held
 652                   : waiter->mode == OVSDB_LOCK_WAIT ? &locks_waiting
 653                   : &locks_lost);
 654         if (string->length) {
 655             ds_put_char(string, ' ');
 656         }
 657         ds_put_cstr(string, waiter->lock_name);
 658     }
 659     status->locks_held = ds_steal_cstr(&locks_held);
 660     status->locks_waiting = ds_steal_cstr(&locks_waiting);
 661     status->locks_lost = ds_steal_cstr(&locks_lost);
 662 }
 663
 664 /* Examines 'request' to determine the database to which it relates, and then
 665  * searches 's' to find that database:
 666  *
 667  *    - If successful, returns the database and sets '*replyp' to NULL.
 668  *
 669  *    - If no such database exists, returns NULL and sets '*replyp' to an
 670  *      appropriate JSON-RPC error reply, owned by the caller. */
 671 static struct ovsdb *
 672 ovsdb_jsonrpc_lookup_db(const struct ovsdb_jsonrpc_session *s,
 673                         const struct jsonrpc_msg *request,
 674                         struct jsonrpc_msg **replyp)
 675 {
 676     struct json_array *params;
 677     struct ovsdb_error *error;
 678     const char *db_name;
 679     struct ovsdb *db;
 680
 681     params = json_array(request->params);
 682     if (!params->n || params->elems[0]->type != JSON_STRING) {
 683         error = ovsdb_syntax_error(
 684             request->params, NULL,
 685             "%s request params must begin with <db-name>", request->method);
 686         goto error;
 687     }
 688
 689     db_name = params->elems[0]->u.string;
 690     db = shash_find_data(&s->up.server->dbs, db_name);
 691     if (!db) {
 692         error = ovsdb_syntax_error(
 693             request->params, "unknown database",
 694             "%s request specifies unknown database %s",
 695             request->method, db_name);
 696         goto error;
 697     }
 698
 699     *replyp = NULL;
 700     return db;
 701
 702 error:
 703     *replyp = jsonrpc_create_error(ovsdb_error_to_json(error), request->id);
 704     ovsdb_error_destroy(error);
 705     return NULL;
 706 }
 707
 708 static struct ovsdb_error *
 709 ovsdb_jsonrpc_session_parse_lock_name(const struct jsonrpc_msg *request,
 710                                       const char **lock_namep)
 711 {
 712     const struct json_array *params;
 713
 714     params = json_array(request->params);
 715     if (params->n != 1 || params->elems[0]->type != JSON_STRING ||
 716         !ovsdb_parser_is_id(json_string(params->elems[0]))) {
 717         *lock_namep = NULL;
 718         return ovsdb_syntax_error(request->params, NULL,
 719                                   "%s request params must be <id>",
 720                                   request->method);
 721     }
 722
 723     *lock_namep = json_string(params->elems[0]);
 724     return NULL;
 725 }
 726
 727 static void
 728 ovsdb_jsonrpc_session_notify(struct ovsdb_session *session,
 729                              const char *lock_name,
 730                              const char *method)
 731 {
 732     struct ovsdb_jsonrpc_session *s;
 733     struct json *params;
 734
 735     s = CONTAINER_OF(session, struct ovsdb_jsonrpc_session, up);
 736     params = json_array_create_1(json_string_create(lock_name));
 737     ovsdb_jsonrpc_session_send(s, jsonrpc_create_notify(method, params));
 738 }
 739
 740 static struct jsonrpc_msg *
 741 ovsdb_jsonrpc_session_lock(struct ovsdb_jsonrpc_session *s,
 742                            struct jsonrpc_msg *request,
 743                            enum ovsdb_lock_mode mode)
 744 {
 745     struct ovsdb_lock_waiter *waiter;
 746     struct jsonrpc_msg *reply;
 747     struct ovsdb_error *error;
 748     struct ovsdb_session *victim;
 749     const char *lock_name;
 750     struct json *result;
 751
 752     error = ovsdb_jsonrpc_session_parse_lock_name(request, &lock_name);
 753     if (error) {
 754         goto error;
 755     }
 756
 757     /* Report error if this session has issued a "lock" or "steal" without a
 758      * matching "unlock" for this lock. */
 759     waiter = ovsdb_session_get_lock_waiter(&s->up, lock_name);
 760     if (waiter) {
 761         error = ovsdb_syntax_error(
 762             request->params, NULL,
 763             "must issue \"unlock\" before new \"%s\"", request->method);
 764         goto error;
 765     }
 766
 767     /* Get the lock, add us as a waiter. */
 768     waiter = ovsdb_server_lock(&s->remote->server->up, &s->up, lock_name, mode,
 769                                &victim);
 770     if (victim) {
 771         ovsdb_jsonrpc_session_notify(victim, lock_name, "stolen");
 772     }
 773
 774     result = json_object_create();
 775     json_object_put(result, "locked",
 776                     json_boolean_create(ovsdb_lock_waiter_is_owner(waiter)));
 777
 778     return jsonrpc_create_reply(result, request->id);
 779
 780 error:
 781     reply = jsonrpc_create_error(ovsdb_error_to_json(error), request->id);
 782     ovsdb_error_destroy(error);
 783     return reply;
 784 }
 785
 786 static void
 787 ovsdb_jsonrpc_session_unlock_all(struct ovsdb_jsonrpc_session *s)
 788 {
 789     struct ovsdb_lock_waiter *waiter, *next;
 790
 791     HMAP_FOR_EACH_SAFE (waiter, next, session_node, &s->up.waiters) {
 792         ovsdb_jsonrpc_session_unlock__(waiter);
 793     }
 794 }
 795
 796 static void
 797 ovsdb_jsonrpc_session_unlock__(struct ovsdb_lock_waiter *waiter)
 798 {
 799     struct ovsdb_lock *lock = waiter->lock;
 800
 801     if (lock) {
 802         struct ovsdb_session *new_owner = ovsdb_lock_waiter_remove(waiter);
 803         if (new_owner) {
 804             ovsdb_jsonrpc_session_notify(new_owner, lock->name, "locked");
 805         } else {
 806             /* ovsdb_server_lock() might have freed 'lock'. */
 807         }
 808     }
 809
 810     ovsdb_lock_waiter_destroy(waiter);
 811 }
 812
 813 static struct jsonrpc_msg *
 814 ovsdb_jsonrpc_session_unlock(struct ovsdb_jsonrpc_session *s,
 815                              struct jsonrpc_msg *request)
 816 {
 817     struct ovsdb_lock_waiter *waiter;
 818     struct jsonrpc_msg *reply;
 819     struct ovsdb_error *error;
 820     const char *lock_name;
 821
 822     error = ovsdb_jsonrpc_session_parse_lock_name(request, &lock_name);
 823     if (error) {
 824         goto error;
 825     }
 826
 827     /* Report error if this session has not issued a "lock" or "steal" for this
 828      * lock. */
 829     waiter = ovsdb_session_get_lock_waiter(&s->up, lock_name);
 830     if (!waiter) {
 831         error = ovsdb_syntax_error(
 832             request->params, NULL, "\"unlock\" without \"lock\" or \"steal\"");
 833         goto error;
 834     }
 835
 836     ovsdb_jsonrpc_session_unlock__(waiter);
 837
 838     return jsonrpc_create_reply(json_object_create(), request->id);
 839
 840 error:
 841     reply = jsonrpc_create_error(ovsdb_error_to_json(error), request->id);
 842     ovsdb_error_destroy(error);
 843     return reply;
 844 }
 845
 846 static struct jsonrpc_msg *
 847 execute_transaction(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
 848                     struct jsonrpc_msg *request)
 849 {
 850     ovsdb_jsonrpc_trigger_create(s, db, request->id, request->params);
 851     request->id = NULL;
 852     request->params = NULL;
 853     jsonrpc_msg_destroy(request);
 854     return NULL;
 855 }
 856
 857 static void
 858 ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *s,
 859                                   struct jsonrpc_msg *request)
 860 {
 861     struct jsonrpc_msg *reply;
 862
 863     if (!strcmp(request->method, "transact")) {
 864         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
 865         if (!reply) {
 866             reply = execute_transaction(s, db, request);
 867         }
 868     } else if (!strcmp(request->method, "monitor") ||
 869                (monitor2_enable__ && !strcmp(request->method, "monitor2")) ||
 870                !strcmp(request->method, "monitor_cond")) {
 871         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
 872         if (!reply) {
 873             int l = strlen(request->method) - strlen("monitor");
 874             enum ovsdb_monitor_version version = l ? OVSDB_MONITOR_V2
 875                                                    : OVSDB_MONITOR_V1;
 876             reply = ovsdb_jsonrpc_monitor_create(s, db, request->params,
 877                                                  version, request->id);
 878         }
 879     } else if (!strcmp(request->method, "monitor_cancel")) {
 880         reply = ovsdb_jsonrpc_monitor_cancel(s, json_array(request->params),
 881                                              request->id);
 882     } else if (!strcmp(request->method, "get_schema")) {
 883         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
 884         if (!reply) {
 885             reply = jsonrpc_create_reply(ovsdb_schema_to_json(db->schema),
 886                                          request->id);
 887         }
 888     } else if (!strcmp(request->method, "list_dbs")) {
 889         size_t n_dbs = shash_count(&s->up.server->dbs);
 890         struct shash_node *node;
 891         struct json **dbs;
 892         size_t i;
 893
 894         dbs = xmalloc(n_dbs * sizeof *dbs);
 895         i = 0;
 896         SHASH_FOR_EACH (node, &s->up.server->dbs) {
 897             dbs[i++] = json_string_create(node->name);
 898         }
 899         reply = jsonrpc_create_reply(json_array_create(dbs, n_dbs),
 900                                      request->id);
 901     } else if (!strcmp(request->method, "lock")) {
 902         reply = ovsdb_jsonrpc_session_lock(s, request, OVSDB_LOCK_WAIT);
 903     } else if (!strcmp(request->method, "steal")) {
 904         reply = ovsdb_jsonrpc_session_lock(s, request, OVSDB_LOCK_STEAL);
 905     } else if (!strcmp(request->method, "unlock")) {
 906         reply = ovsdb_jsonrpc_session_unlock(s, request);
 907     } else if (!strcmp(request->method, "echo")) {
 908         reply = jsonrpc_create_reply(json_clone(request->params), request->id);
 909     } else {
 910         reply = jsonrpc_create_error(json_string_create("unknown method"),
 911                                      request->id);
 912     }
 913
 914     if (reply) {
 915         jsonrpc_msg_destroy(request);
 916         ovsdb_jsonrpc_session_send(s, reply);
 917     }
 918 }
 919
 920 static void
 921 execute_cancel(struct ovsdb_jsonrpc_session *s, struct jsonrpc_msg *request)
 922 {
 923     if (json_array(request->params)->n == 1) {
 924         struct ovsdb_jsonrpc_trigger *t;
 925         struct json *id;
 926
 927         id = request->params->u.array.elems[0];
 928         t = ovsdb_jsonrpc_trigger_find(s, id, json_hash(id, 0));
 929         if (t) {
 930             ovsdb_jsonrpc_trigger_complete(t);
 931         }
 932     }
 933 }
 934
 935 static void
 936 ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *s,
 937                                  struct jsonrpc_msg *request)
 938 {
 939     if (!strcmp(request->method, "cancel")) {
 940         execute_cancel(s, request);
 941     }
 942     jsonrpc_msg_destroy(request);
 943 }
 944
 945 static void
 946 ovsdb_jsonrpc_session_send(struct ovsdb_jsonrpc_session *s,
 947                            struct jsonrpc_msg *msg)
 948 {
 949     ovsdb_jsonrpc_monitor_flush_all(s);
 950     jsonrpc_session_send(s->js, msg);
 951 }
 952 \f
 953 /* JSON-RPC database server triggers.
 954  *
 955  * (Every transaction is treated as a trigger even if it doesn't actually have
 956  * any "wait" operations.) */
 957
 958 struct ovsdb_jsonrpc_trigger {
 959     struct ovsdb_trigger trigger;
 960     struct hmap_node hmap_node; /* In session's "triggers" hmap. */
 961     struct json *id;
 962 };
 963
 964 static void
 965 ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
 966                              struct json *id, struct json *params)
 967 {
 968     struct ovsdb_jsonrpc_trigger *t;
 969     size_t hash;
 970
 971     /* Check for duplicate ID. */
 972     hash = json_hash(id, 0);
 973     t = ovsdb_jsonrpc_trigger_find(s, id, hash);
 974     if (t) {
 975         struct jsonrpc_msg *msg;
 976
 977         msg = jsonrpc_create_error(json_string_create("duplicate request ID"),
 978                                    id);
 979         ovsdb_jsonrpc_session_send(s, msg);
 980         json_destroy(id);
 981         json_destroy(params);
 982         return;
 983     }
 984
 985     /* Insert into trigger table. */
 986     t = xmalloc(sizeof *t);
 987     ovsdb_trigger_init(&s->up, db, &t->trigger, params, time_msec());
 988     t->id = id;
 989     hmap_insert(&s->triggers, &t->hmap_node, hash);
 990
 991     /* Complete early if possible. */
 992     if (ovsdb_trigger_is_complete(&t->trigger)) {
 993         ovsdb_jsonrpc_trigger_complete(t);
 994     }
 995 }
 996
 997 static struct ovsdb_jsonrpc_trigger *
 998 ovsdb_jsonrpc_trigger_find(struct ovsdb_jsonrpc_session *s,
 999                            const struct json *id, size_t hash)
1000 {
1001     struct ovsdb_jsonrpc_trigger *t;
1002
1003     HMAP_FOR_EACH_WITH_HASH (t, hmap_node, hash, &s->triggers) {
1004         if (json_equal(t->id, id)) {
1005             return t;
1006         }
1007     }
1008
1009     return NULL;
1010 }
1011
1012 static void
1013 ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *t)
1014 {
1015     struct ovsdb_jsonrpc_session *s;
1016
1017     s = CONTAINER_OF(t->trigger.session, struct ovsdb_jsonrpc_session, up);
1018
1019     if (jsonrpc_session_is_connected(s->js)) {
1020         struct jsonrpc_msg *reply;
1021         struct json *result;
1022
1023         result = ovsdb_trigger_steal_result(&t->trigger);
1024         if (result) {
1025             reply = jsonrpc_create_reply(result, t->id);
1026         } else {
1027             reply = jsonrpc_create_error(json_string_create("canceled"),
1028                                          t->id);
1029         }
1030         ovsdb_jsonrpc_session_send(s, reply);
1031     }
1032
1033     json_destroy(t->id);
1034     ovsdb_trigger_destroy(&t->trigger);
1035     hmap_remove(&s->triggers, &t->hmap_node);
1036     free(t);
1037 }
1038
1039 static void
1040 ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *s)
1041 {
1042     struct ovsdb_jsonrpc_trigger *t, *next;
1043     HMAP_FOR_EACH_SAFE (t, next, hmap_node, &s->triggers) {
1044         ovsdb_jsonrpc_trigger_complete(t);
1045     }
1046 }
1047
1048 static void
1049 ovsdb_jsonrpc_trigger_complete_done(struct ovsdb_jsonrpc_session *s)
1050 {
1051     while (!ovs_list_is_empty(&s->up.completions)) {
1052         struct ovsdb_jsonrpc_trigger *t
1053             = CONTAINER_OF(s->up.completions.next,
1054                            struct ovsdb_jsonrpc_trigger, trigger.node);
1055         ovsdb_jsonrpc_trigger_complete(t);
1056     }
1057 }
1058 \f
1059 /* Jsonrpc front end monitor. */
1060 struct ovsdb_jsonrpc_monitor {
1061     struct ovsdb_jsonrpc_session *session;
1062     struct ovsdb *db;
1063     struct hmap_node node;      /* In ovsdb_jsonrpc_session's "monitors". */
1064     struct json *monitor_id;
1065     struct ovsdb_monitor *dbmon;
1066     uint64_t unflushed;         /* The first transaction that has not been
1067                                        flushed to the jsonrpc remote client. */
1068     enum ovsdb_monitor_version version;
1069     struct ovsdb_monitor_session_condition *condition;/* Session's condition */
1070 };
1071
1072 static struct ovsdb_jsonrpc_monitor *
1073 ovsdb_jsonrpc_monitor_find(struct ovsdb_jsonrpc_session *s,
1074                            const struct json *monitor_id)
1075 {
1076     struct ovsdb_jsonrpc_monitor *m;
1077
1078     HMAP_FOR_EACH_WITH_HASH (m, node, json_hash(monitor_id, 0), &s->monitors) {
1079         if (json_equal(m->monitor_id, monitor_id)) {
1080             return m;
1081         }
1082     }
1083
1084     return NULL;
1085 }
1086
1087 static bool
1088 parse_bool(struct ovsdb_parser *parser, const char *name, bool default_value)
1089 {
1090     const struct json *json;
1091
1092     json = ovsdb_parser_member(parser, name, OP_BOOLEAN | OP_OPTIONAL);
1093     return json ? json_boolean(json) : default_value;
1094 }
1095
1096 static struct ovsdb_error * OVS_WARN_UNUSED_RESULT
1097 ovsdb_jsonrpc_parse_monitor_request(
1098                                struct ovsdb_monitor *dbmon,
1099                                const struct ovsdb_table *table,
1100                                struct ovsdb_monitor_session_condition *cond,
1101                                const struct json *monitor_request)
1102 {
1103     const struct ovsdb_table_schema *ts = table->schema;
1104     enum ovsdb_monitor_selection select;
1105     const struct json *columns, *select_json, *where = NULL;
1106     struct ovsdb_parser parser;
1107     struct ovsdb_error *error;
1108
1109     ovsdb_parser_init(&parser, monitor_request, "table %s", ts->name);
1110     if (cond) {
1111         where = ovsdb_parser_member(&parser, "where", OP_ARRAY | OP_OPTIONAL);
1112     }
1113     columns = ovsdb_parser_member(&parser, "columns", OP_ARRAY | OP_OPTIONAL);
1114
1115     select_json = ovsdb_parser_member(&parser, "select",
1116                                       OP_OBJECT | OP_OPTIONAL);
1117
1118     error = ovsdb_parser_finish(&parser);
1119     if (error) {
1120         return error;
1121     }
1122
1123     if (select_json) {
1124         select = 0;
1125         ovsdb_parser_init(&parser, select_json, "table %s select", ts->name);
1126         if (parse_bool(&parser, "initial", true)) {
1127             select |= OJMS_INITIAL;
1128         }
1129         if (parse_bool(&parser, "insert", true)) {
1130             select |= OJMS_INSERT;
1131         }
1132         if (parse_bool(&parser, "delete", true)) {
1133             select |= OJMS_DELETE;
1134         }
1135         if (parse_bool(&parser, "modify", true)) {
1136             select |= OJMS_MODIFY;
1137         }
1138         error = ovsdb_parser_finish(&parser);
1139         if (error) {
1140             return error;
1141         }
1142     } else {
1143         select = OJMS_INITIAL | OJMS_INSERT | OJMS_DELETE | OJMS_MODIFY;
1144     }
1145
1146     ovsdb_monitor_table_add_select(dbmon, table, select);
1147     if (columns) {
1148         size_t i;
1149
1150         if (columns->type != JSON_ARRAY) {
1151             return ovsdb_syntax_error(columns, NULL,
1152                                       "array of column names expected");
1153         }
1154
1155         for (i = 0; i < columns->u.array.n; i++) {
1156             const struct ovsdb_column *column;
1157             const char *s;
1158
1159             if (columns->u.array.elems[i]->type != JSON_STRING) {
1160                 return ovsdb_syntax_error(columns, NULL,
1161                                           "array of column names expected");
1162             }
1163
1164             s = columns->u.array.elems[i]->u.string;
1165             column = shash_find_data(&table->schema->columns, s);
1166             if (!column) {
1167                 return ovsdb_syntax_error(columns, NULL, "%s is not a valid "
1168                                           "column name", s);
1169             }
1170             if (ovsdb_monitor_add_column(dbmon, table, column,
1171                                          select, true)) {
1172                 return ovsdb_syntax_error(columns, NULL, "column %s "
1173                                           "mentioned more than once",
1174                                           column->name);
1175             }
1176         }
1177     } else {
1178         struct shash_node *node;
1179
1180         SHASH_FOR_EACH (node, &ts->columns) {
1181             const struct ovsdb_column *column = node->data;
1182             if (column->index != OVSDB_COL_UUID) {
1183                 if (ovsdb_monitor_add_column(dbmon, table, column,
1184                                              select, true)) {
1185                     return ovsdb_syntax_error(columns, NULL, "column %s "
1186                                               "mentioned more than once",
1187                                               column->name);
1188                 }
1189             }
1190         }
1191     }
1192     if (cond) {
1193         error = ovsdb_monitor_table_condition_create(cond, table, where);
1194         if (error) {
1195             return error;
1196         }
1197     }
1198
1199     return NULL;
1200 }
1201
1202 static struct jsonrpc_msg *
1203 ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
1204                              struct json *params,
1205                              enum ovsdb_monitor_version version,
1206                              const struct json *request_id)
1207 {
1208     struct ovsdb_jsonrpc_monitor *m = NULL;
1209     struct ovsdb_monitor *dbmon = NULL;
1210     struct json *monitor_id, *monitor_requests;
1211     struct ovsdb_error *error = NULL;
1212     struct shash_node *node;
1213     struct json *json;
1214
1215     if (json_array(params)->n != 3) {
1216         error = ovsdb_syntax_error(params, NULL, "invalid parameters");
1217         goto error;
1218     }
1219     monitor_id = params->u.array.elems[1];
1220     monitor_requests = params->u.array.elems[2];
1221     if (monitor_requests->type != JSON_OBJECT) {
1222         error = ovsdb_syntax_error(monitor_requests, NULL,
1223                                    "monitor-requests must be object");
1224         goto error;
1225     }
1226
1227     if (ovsdb_jsonrpc_monitor_find(s, monitor_id)) {
1228         error = ovsdb_syntax_error(monitor_id, NULL, "duplicate monitor ID");
1229         goto error;
1230     }
1231
1232     m = xzalloc(sizeof *m);
1233     m->session = s;
1234     m->db = db;
1235     m->dbmon = ovsdb_monitor_create(db, m);
1236     if (version == OVSDB_MONITOR_V2) {
1237         m->condition = ovsdb_monitor_session_condition_create();
1238     }
1239     m->unflushed = 0;
1240     m->version = version;
1241     hmap_insert(&s->monitors, &m->node, json_hash(monitor_id, 0));
1242     m->monitor_id = json_clone(monitor_id);
1243
1244     SHASH_FOR_EACH (node, json_object(monitor_requests)) {
1245         const struct ovsdb_table *table;
1246         const struct json *mr_value;
1247         size_t i;
1248
1249         table = ovsdb_get_table(m->db, node->name);
1250         if (!table) {
1251             error = ovsdb_syntax_error(NULL, NULL,
1252                                        "no table named %s", node->name);
1253             goto error;
1254         }
1255
1256         ovsdb_monitor_add_table(m->dbmon, table);
1257
1258         /* Parse columns. */
1259         mr_value = node->data;
1260         if (mr_value->type == JSON_ARRAY) {
1261             const struct json_array *array = &mr_value->u.array;
1262
1263             for (i = 0; i < array->n; i++) {
1264                 error = ovsdb_jsonrpc_parse_monitor_request(m->dbmon,
1265                                                             table,
1266                                                             m->condition,
1267                                                             array->elems[i]);
1268                 if (error) {
1269                     goto error;
1270                 }
1271             }
1272         } else {
1273             error = ovsdb_jsonrpc_parse_monitor_request(m->dbmon,
1274                                                         table,
1275                                                         m->condition,
1276                                                         mr_value);
1277             if (error) {
1278                 goto error;
1279             }
1280         }
1281     }
1282
1283     dbmon = ovsdb_monitor_add(m->dbmon);
1284     if (dbmon != m->dbmon) {
1285         /* Found an exisiting dbmon, reuse the current one. */
1286         ovsdb_monitor_remove_jsonrpc_monitor(m->dbmon, m, m->unflushed);
1287         ovsdb_monitor_add_jsonrpc_monitor(dbmon, m);
1288         m->dbmon = dbmon;
1289     }
1290
1291     /* Only now we can bind session's condition to ovsdb_monitor */
1292     if (m->condition) {
1293         ovsdb_monitor_condition_bind(m->dbmon, m->condition);
1294     }
1295
1296     ovsdb_monitor_get_initial(m->dbmon);
1297     json = ovsdb_jsonrpc_monitor_compose_update(m, true);
1298     json = json ? json : json_object_create();
1299     return jsonrpc_create_reply(json, request_id);
1300
1301 error:
1302     if (m) {
1303         ovsdb_jsonrpc_monitor_destroy(m);
1304     }
1305
1306     json = ovsdb_error_to_json(error);
1307     ovsdb_error_destroy(error);
1308     return jsonrpc_create_error(json, request_id);
1309 }
1310
1311 static struct jsonrpc_msg *
1312 ovsdb_jsonrpc_monitor_cancel(struct ovsdb_jsonrpc_session *s,
1313                              struct json_array *params,
1314                              const struct json *request_id)
1315 {
1316     if (params->n != 1) {
1317         return jsonrpc_create_error(json_string_create("invalid parameters"),
1318                                     request_id);
1319     } else {
1320         struct ovsdb_jsonrpc_monitor *m;
1321
1322         m = ovsdb_jsonrpc_monitor_find(s, params->elems[0]);
1323         if (!m) {
1324             return jsonrpc_create_error(json_string_create("unknown monitor"),
1325                                         request_id);
1326         } else {
1327             ovsdb_jsonrpc_monitor_destroy(m);
1328             return jsonrpc_create_reply(json_object_create(), request_id);
1329         }
1330     }
1331 }
1332
1333 static void
1334 ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *s)
1335 {
1336     struct ovsdb_jsonrpc_monitor *m, *next;
1337
1338     HMAP_FOR_EACH_SAFE (m, next, node, &s->monitors) {
1339         ovsdb_jsonrpc_monitor_destroy(m);
1340     }
1341 }
1342
1343 static struct json *
1344 ovsdb_jsonrpc_monitor_compose_update(struct ovsdb_jsonrpc_monitor *m,
1345                                      bool initial)
1346 {
1347     if (!ovsdb_monitor_needs_flush(m->dbmon, m->unflushed)) {
1348         return NULL;
1349     }
1350
1351     return ovsdb_monitor_get_update(m->dbmon, initial, &m->unflushed,
1352                                     m->condition, m->version);
1353 }
1354
1355 static bool
1356 ovsdb_jsonrpc_monitor_needs_flush(struct ovsdb_jsonrpc_session *s)
1357 {
1358     struct ovsdb_jsonrpc_monitor *m;
1359
1360     HMAP_FOR_EACH (m, node, &s->monitors) {
1361         if (ovsdb_monitor_needs_flush(m->dbmon, m->unflushed)) {
1362             return true;
1363         }
1364     }
1365
1366     return false;
1367 }
1368
1369 void
1370 ovsdb_jsonrpc_monitor_destroy(struct ovsdb_jsonrpc_monitor *m)
1371 {
1372     json_destroy(m->monitor_id);
1373     hmap_remove(&m->session->monitors, &m->node);
1374     ovsdb_monitor_remove_jsonrpc_monitor(m->dbmon, m, m->unflushed);
1375     ovsdb_monitor_session_condition_destroy(m->condition);
1376     free(m);
1377 }
1378
1379 static struct jsonrpc_msg *
1380 ovsdb_jsonrpc_create_notify(const struct ovsdb_jsonrpc_monitor *m,
1381                             struct json *params)
1382 {
1383     const char *method;
1384
1385     switch(m->version) {
1386     case OVSDB_MONITOR_V1:
1387         method = "update";
1388         break;
1389     case OVSDB_MONITOR_V2:
1390         method = "update2";
1391         break;
1392     case OVSDB_MONITOR_VERSION_MAX:
1393     default:
1394         OVS_NOT_REACHED();
1395     }
1396
1397     return jsonrpc_create_notify(method, params);
1398 }
1399
1400 static void
1401 ovsdb_jsonrpc_monitor_flush_all(struct ovsdb_jsonrpc_session *s)
1402 {
1403     struct ovsdb_jsonrpc_monitor *m;
1404
1405     HMAP_FOR_EACH (m, node, &s->monitors) {
1406         struct json *json;
1407
1408         json = ovsdb_jsonrpc_monitor_compose_update(m, false);
1409         if (json) {
1410             struct jsonrpc_msg *msg;
1411             struct json *params;
1412
1413             params = json_array_create_2(json_clone(m->monitor_id), json);
1414             msg = ovsdb_jsonrpc_create_notify(m, params);
1415             jsonrpc_session_send(s->js, msg);
1416         }
1417     }
1418 }
1419
1420 void
1421 ovsdb_jsonrpc_disable_monitor2(void)
1422 {
1423     /* Once disabled, it is not possible to re-enable it. */
1424     monitor2_enable__ = false;
1425 }