ovsdb/jsonrpc-server.c

   1 /* Copyright (c) 2009, 2010 Nicira Networks
   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 "column.h"
  23 #include "json.h"
  24 #include "jsonrpc.h"
  25 #include "ovsdb-error.h"
  26 #include "ovsdb-parser.h"
  27 #include "ovsdb.h"
  28 #include "reconnect.h"
  29 #include "row.h"
  30 #include "stream.h"
  31 #include "table.h"
  32 #include "timeval.h"
  33 #include "transaction.h"
  34 #include "trigger.h"
  35
  36 #define THIS_MODULE VLM_ovsdb_jsonrpc_server
  37 #include "vlog.h"
  38
  39 struct ovsdb_jsonrpc_remote;
  40 struct ovsdb_jsonrpc_session;
  41
  42 /* Message rate-limiting. */
  43 struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
  44
  45 /* Sessions. */
  46 static struct ovsdb_jsonrpc_session *ovsdb_jsonrpc_session_create(
  47     struct ovsdb_jsonrpc_remote *, struct jsonrpc_session *);
  48 static void ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_remote *);
  49 static void ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_remote *);
  50 static void ovsdb_jsonrpc_session_close_all(struct ovsdb_jsonrpc_remote *);
  51
  52 /* Triggers. */
  53 static void ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *,
  54                                          struct json *id, struct json *params);
  55 static struct ovsdb_jsonrpc_trigger *ovsdb_jsonrpc_trigger_find(
  56     struct ovsdb_jsonrpc_session *, const struct json *id, size_t hash);
  57 static void ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *);
  58 static void ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *);
  59 static void ovsdb_jsonrpc_trigger_complete_done(
  60     struct ovsdb_jsonrpc_session *);
  61
  62 /* Monitors. */
  63 static struct json *ovsdb_jsonrpc_monitor_create(
  64     struct ovsdb_jsonrpc_session *, struct json *params);
  65 static struct jsonrpc_msg *ovsdb_jsonrpc_monitor_cancel(
  66     struct ovsdb_jsonrpc_session *,
  67     struct json_array *params,
  68     const struct json *request_id);
  69 static void ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *);
  70 \f
  71 /* JSON-RPC database server. */
  72
  73 struct ovsdb_jsonrpc_server {
  74     struct ovsdb *db;
  75     unsigned int n_sessions, max_sessions;
  76     struct shash remotes;      /* Contains "struct ovsdb_jsonrpc_remote *"s. */
  77 };
  78
  79 /* A configured remote.  This is either a passive stream listener plus a list
  80  * of the currently connected sessions, or a list of exactly one active
  81  * session. */
  82 struct ovsdb_jsonrpc_remote {
  83     struct ovsdb_jsonrpc_server *server;
  84     struct pstream *listener;   /* Listener, if passive. */
  85     struct list sessions;       /* List of "struct ovsdb_jsonrpc_session"s. */
  86 };
  87
  88 static void ovsdb_jsonrpc_server_add_remote(struct ovsdb_jsonrpc_server *,
  89                                             const char *name);
  90 static void ovsdb_jsonrpc_server_del_remote(struct shash_node *);
  91
  92 struct ovsdb_jsonrpc_server *
  93 ovsdb_jsonrpc_server_create(struct ovsdb *db)
  94 {
  95     struct ovsdb_jsonrpc_server *server = xzalloc(sizeof *server);
  96     server->db = db;
  97     server->max_sessions = 64;
  98     shash_init(&server->remotes);
  99     return server;
 100 }
 101
 102 /* Sets 'svr''s current set of remotes to the names in 'new_remotes'.  The data
 103  * values in 'new_remotes' are ignored.
 104  *
 105  * A remote is an active or passive stream connection method, e.g. "pssl:" or
 106  * "tcp:1.2.3.4". */
 107 void
 108 ovsdb_jsonrpc_server_set_remotes(struct ovsdb_jsonrpc_server *svr,
 109                                  const struct shash *new_remotes)
 110 {
 111     struct shash_node *node, *next;
 112
 113     SHASH_FOR_EACH_SAFE (node, next, &svr->remotes) {
 114         if (!shash_find(new_remotes, node->name)) {
 115             ovsdb_jsonrpc_server_del_remote(node);
 116         }
 117     }
 118     SHASH_FOR_EACH (node, new_remotes) {
 119         if (!shash_find(&svr->remotes, node->name)) {
 120             ovsdb_jsonrpc_server_add_remote(svr, node->name);
 121         }
 122     }
 123 }
 124
 125 static void
 126 ovsdb_jsonrpc_server_add_remote(struct ovsdb_jsonrpc_server *svr,
 127                                 const char *name)
 128 {
 129     struct ovsdb_jsonrpc_remote *remote;
 130     struct pstream *listener;
 131     int error;
 132
 133     error = pstream_open(name, &listener);
 134     if (error && error != EAFNOSUPPORT) {
 135         VLOG_ERR_RL(&rl, "%s: listen failed: %s", name, strerror(error));
 136         return;
 137     }
 138
 139     remote = xmalloc(sizeof *remote);
 140     remote->server = svr;
 141     remote->listener = listener;
 142     list_init(&remote->sessions);
 143     shash_add(&svr->remotes, name, remote);
 144
 145     if (!listener) {
 146         ovsdb_jsonrpc_session_create(remote, jsonrpc_session_open(name));
 147     }
 148 }
 149
 150 static void
 151 ovsdb_jsonrpc_server_del_remote(struct shash_node *node)
 152 {
 153     struct ovsdb_jsonrpc_remote *remote = node->data;
 154
 155     ovsdb_jsonrpc_session_close_all(remote);
 156     pstream_close(remote->listener);
 157     shash_delete(&remote->server->remotes, node);
 158     free(remote);
 159 }
 160
 161 void
 162 ovsdb_jsonrpc_server_run(struct ovsdb_jsonrpc_server *svr)
 163 {
 164     struct shash_node *node;
 165
 166     SHASH_FOR_EACH (node, &svr->remotes) {
 167         struct ovsdb_jsonrpc_remote *remote = node->data;
 168
 169         if (remote->listener && svr->n_sessions < svr->max_sessions) {
 170             struct stream *stream;
 171             int error;
 172
 173             error = pstream_accept(remote->listener, &stream);
 174             if (!error) {
 175                 struct jsonrpc_session *js;
 176                 js = jsonrpc_session_open_unreliably(jsonrpc_open(stream));
 177                 ovsdb_jsonrpc_session_create(remote, js);
 178             } else if (error != EAGAIN) {
 179                 VLOG_WARN_RL(&rl, "%s: accept failed: %s",
 180                              pstream_get_name(remote->listener),
 181                              strerror(error));
 182             }
 183         }
 184
 185         ovsdb_jsonrpc_session_run_all(remote);
 186     }
 187 }
 188
 189 void
 190 ovsdb_jsonrpc_server_wait(struct ovsdb_jsonrpc_server *svr)
 191 {
 192     struct shash_node *node;
 193
 194     SHASH_FOR_EACH (node, &svr->remotes) {
 195         struct ovsdb_jsonrpc_remote *remote = node->data;
 196
 197         if (remote->listener && svr->n_sessions < svr->max_sessions) {
 198             pstream_wait(remote->listener);
 199         }
 200
 201         ovsdb_jsonrpc_session_wait_all(remote);
 202     }
 203 }
 204 \f
 205 /* JSON-RPC database server session. */
 206
 207 struct ovsdb_jsonrpc_session {
 208     struct ovsdb_jsonrpc_remote *remote;
 209     struct list node;           /* Element in remote's sessions list. */
 210
 211     /* Triggers. */
 212     struct hmap triggers;       /* Hmap of "struct ovsdb_jsonrpc_trigger"s. */
 213     struct list completions;    /* Completed triggers. */
 214
 215     /* Monitors. */
 216     struct hmap monitors;       /* Hmap of "struct ovsdb_jsonrpc_monitor"s. */
 217
 218     /* Network connectivity. */
 219     struct jsonrpc_session *js;  /* JSON-RPC session. */
 220     unsigned int js_seqno;       /* Last jsonrpc_session_get_seqno() value. */
 221 };
 222
 223 static void ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *);
 224 static int ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *);
 225 static void ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *);
 226 static void ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *,
 227                                              struct jsonrpc_msg *);
 228 static void ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *,
 229                                              struct jsonrpc_msg *);
 230
 231 static struct ovsdb_jsonrpc_session *
 232 ovsdb_jsonrpc_session_create(struct ovsdb_jsonrpc_remote *remote,
 233                              struct jsonrpc_session *js)
 234 {
 235     struct ovsdb_jsonrpc_session *s;
 236
 237     s = xzalloc(sizeof *s);
 238     s->remote = remote;
 239     list_push_back(&remote->sessions, &s->node);
 240     hmap_init(&s->triggers);
 241     hmap_init(&s->monitors);
 242     list_init(&s->completions);
 243     s->js = js;
 244     s->js_seqno = jsonrpc_session_get_seqno(js);
 245
 246     remote->server->n_sessions++;
 247
 248     return s;
 249 }
 250
 251 static void
 252 ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *s)
 253 {
 254     jsonrpc_session_close(s->js);
 255     list_remove(&s->node);
 256     s->remote->server->n_sessions--;
 257 }
 258
 259 static int
 260 ovsdb_jsonrpc_session_run(struct ovsdb_jsonrpc_session *s)
 261 {
 262     jsonrpc_session_run(s->js);
 263     if (s->js_seqno != jsonrpc_session_get_seqno(s->js)) {
 264         s->js_seqno = jsonrpc_session_get_seqno(s->js);
 265         ovsdb_jsonrpc_trigger_complete_all(s);
 266         ovsdb_jsonrpc_monitor_remove_all(s);
 267     }
 268
 269     ovsdb_jsonrpc_trigger_complete_done(s);
 270
 271     if (!jsonrpc_session_get_backlog(s->js)) {
 272         struct jsonrpc_msg *msg = jsonrpc_session_recv(s->js);
 273         if (msg) {
 274             if (msg->type == JSONRPC_REQUEST) {
 275                 ovsdb_jsonrpc_session_got_request(s, msg);
 276             } else if (msg->type == JSONRPC_NOTIFY) {
 277                 ovsdb_jsonrpc_session_got_notify(s, msg);
 278             } else {
 279                 VLOG_WARN("%s: received unexpected %s message",
 280                           jsonrpc_session_get_name(s->js),
 281                           jsonrpc_msg_type_to_string(msg->type));
 282                 jsonrpc_session_force_reconnect(s->js);
 283                 jsonrpc_msg_destroy(msg);
 284             }
 285         }
 286     }
 287     return jsonrpc_session_is_alive(s->js) ? 0 : ETIMEDOUT;
 288 }
 289
 290 static void
 291 ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_remote *remote)
 292 {
 293     struct ovsdb_jsonrpc_session *s, *next;
 294
 295     LIST_FOR_EACH_SAFE (s, next, struct ovsdb_jsonrpc_session, node,
 296                         &remote->sessions) {
 297         int error = ovsdb_jsonrpc_session_run(s);
 298         if (error) {
 299             ovsdb_jsonrpc_session_close(s);
 300         }
 301     }
 302 }
 303
 304 static void
 305 ovsdb_jsonrpc_session_wait(struct ovsdb_jsonrpc_session *s)
 306 {
 307     jsonrpc_session_wait(s->js);
 308     if (!jsonrpc_session_get_backlog(s->js)) {
 309         jsonrpc_session_recv_wait(s->js);
 310     }
 311 }
 312
 313 static void
 314 ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_remote *remote)
 315 {
 316     struct ovsdb_jsonrpc_session *s;
 317
 318     LIST_FOR_EACH (s, struct ovsdb_jsonrpc_session, node, &remote->sessions) {
 319         ovsdb_jsonrpc_session_wait(s);
 320     }
 321 }
 322
 323 static void
 324 ovsdb_jsonrpc_session_close_all(struct ovsdb_jsonrpc_remote *remote)
 325 {
 326     struct ovsdb_jsonrpc_session *s, *next;
 327
 328     LIST_FOR_EACH_SAFE (s, next, struct ovsdb_jsonrpc_session, node,
 329                         &remote->sessions) {
 330         ovsdb_jsonrpc_session_close(s);
 331     }
 332 }
 333
 334 static struct jsonrpc_msg *
 335 execute_transaction(struct ovsdb_jsonrpc_session *s,
 336                     struct jsonrpc_msg *request)
 337 {
 338     ovsdb_jsonrpc_trigger_create(s, request->id, request->params);
 339     request->id = NULL;
 340     request->params = NULL;
 341     return NULL;
 342 }
 343
 344 static void
 345 ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *s,
 346                                   struct jsonrpc_msg *request)
 347 {
 348     struct jsonrpc_msg *reply;
 349
 350     if (!strcmp(request->method, "transact")) {
 351         reply = execute_transaction(s, request);
 352     } else if (!strcmp(request->method, "monitor")) {
 353         reply = jsonrpc_create_reply(
 354             ovsdb_jsonrpc_monitor_create(s, request->params), request->id);
 355     } else if (!strcmp(request->method, "monitor_cancel")) {
 356         reply = ovsdb_jsonrpc_monitor_cancel(s, json_array(request->params),
 357                                              request->id);
 358     } else if (!strcmp(request->method, "get_schema")) {
 359         reply = jsonrpc_create_reply(
 360             ovsdb_schema_to_json(s->remote->server->db->schema), request->id);
 361     } else if (!strcmp(request->method, "echo")) {
 362         reply = jsonrpc_create_reply(json_clone(request->params), request->id);
 363     } else {
 364         reply = jsonrpc_create_error(json_string_create("unknown method"),
 365                                      request->id);
 366     }
 367
 368     if (reply) {
 369         jsonrpc_msg_destroy(request);
 370         jsonrpc_session_send(s->js, reply);
 371     }
 372 }
 373
 374 static void
 375 execute_cancel(struct ovsdb_jsonrpc_session *s, struct jsonrpc_msg *request)
 376 {
 377     if (json_array(request->params)->n == 1) {
 378         struct ovsdb_jsonrpc_trigger *t;
 379         struct json *id;
 380
 381         id = request->params->u.array.elems[0];
 382         t = ovsdb_jsonrpc_trigger_find(s, id, json_hash(id, 0));
 383         if (t) {
 384             ovsdb_jsonrpc_trigger_complete(t);
 385         }
 386     }
 387 }
 388
 389 static void
 390 ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *s,
 391                                  struct jsonrpc_msg *request)
 392 {
 393     if (!strcmp(request->method, "cancel")) {
 394         execute_cancel(s, request);
 395     }
 396     jsonrpc_msg_destroy(request);
 397 }
 398 \f
 399 /* JSON-RPC database server triggers.
 400  *
 401  * (Every transaction is treated as a trigger even if it doesn't actually have
 402  * any "wait" operations.) */
 403
 404 struct ovsdb_jsonrpc_trigger {
 405     struct ovsdb_trigger trigger;
 406     struct ovsdb_jsonrpc_session *session;
 407     struct hmap_node hmap_node; /* In session's "triggers" hmap. */
 408     struct json *id;
 409 };
 410
 411 static void
 412 ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *s,
 413                              struct json *id, struct json *params)
 414 {
 415     struct ovsdb_jsonrpc_trigger *t;
 416     size_t hash;
 417
 418     /* Check for duplicate ID. */
 419     hash = json_hash(id, 0);
 420     t = ovsdb_jsonrpc_trigger_find(s, id, hash);
 421     if (t) {
 422         struct jsonrpc_msg *msg;
 423
 424         msg = jsonrpc_create_error(json_string_create("duplicate request ID"),
 425                                    id);
 426         jsonrpc_session_send(s->js, msg);
 427         json_destroy(id);
 428         json_destroy(params);
 429         return;
 430     }
 431
 432     /* Insert into trigger table. */
 433     t = xmalloc(sizeof *t);
 434     ovsdb_trigger_init(s->remote->server->db,
 435                        &t->trigger, params, &s->completions,
 436                        time_msec());
 437     t->session = s;
 438     t->id = id;
 439     hmap_insert(&s->triggers, &t->hmap_node, hash);
 440
 441     /* Complete early if possible. */
 442     if (ovsdb_trigger_is_complete(&t->trigger)) {
 443         ovsdb_jsonrpc_trigger_complete(t);
 444     }
 445 }
 446
 447 static struct ovsdb_jsonrpc_trigger *
 448 ovsdb_jsonrpc_trigger_find(struct ovsdb_jsonrpc_session *s,
 449                            const struct json *id, size_t hash)
 450 {
 451     struct ovsdb_jsonrpc_trigger *t;
 452
 453     HMAP_FOR_EACH_WITH_HASH (t, struct ovsdb_jsonrpc_trigger, hmap_node, hash,
 454                              &s->triggers) {
 455         if (json_equal(t->id, id)) {
 456             return t;
 457         }
 458     }
 459
 460     return NULL;
 461 }
 462
 463 static void
 464 ovsdb_jsonrpc_trigger_complete(struct ovsdb_jsonrpc_trigger *t)
 465 {
 466     struct ovsdb_jsonrpc_session *s = t->session;
 467
 468     if (jsonrpc_session_is_connected(s->js)) {
 469         struct jsonrpc_msg *reply;
 470         struct json *result;
 471
 472         result = ovsdb_trigger_steal_result(&t->trigger);
 473         if (result) {
 474             reply = jsonrpc_create_reply(result, t->id);
 475         } else {
 476             reply = jsonrpc_create_error(json_string_create("canceled"),
 477                                          t->id);
 478         }
 479         jsonrpc_session_send(s->js, reply);
 480     }
 481
 482     json_destroy(t->id);
 483     ovsdb_trigger_destroy(&t->trigger);
 484     hmap_remove(&s->triggers, &t->hmap_node);
 485     free(t);
 486 }
 487
 488 static void
 489 ovsdb_jsonrpc_trigger_complete_all(struct ovsdb_jsonrpc_session *s)
 490 {
 491     struct ovsdb_jsonrpc_trigger *t, *next;
 492     HMAP_FOR_EACH_SAFE (t, next, struct ovsdb_jsonrpc_trigger, hmap_node,
 493                         &s->triggers) {
 494         ovsdb_jsonrpc_trigger_complete(t);
 495     }
 496 }
 497
 498 static void
 499 ovsdb_jsonrpc_trigger_complete_done(struct ovsdb_jsonrpc_session *s)
 500 {
 501     while (!list_is_empty(&s->completions)) {
 502         struct ovsdb_jsonrpc_trigger *t
 503             = CONTAINER_OF(s->completions.next,
 504                            struct ovsdb_jsonrpc_trigger, trigger.node);
 505         ovsdb_jsonrpc_trigger_complete(t);
 506     }
 507 }
 508 \f
 509 /* JSON-RPC database table monitors. */
 510
 511 enum ovsdb_jsonrpc_monitor_selection {
 512     OJMS_INITIAL = 1 << 0,      /* All rows when monitor is created. */
 513     OJMS_INSERT = 1 << 1,       /* New rows. */
 514     OJMS_DELETE = 1 << 2,       /* Deleted rows. */
 515     OJMS_MODIFY = 1 << 3        /* Modified rows. */
 516 };
 517
 518 struct ovsdb_jsonrpc_monitor_table {
 519     const struct ovsdb_table *table;
 520     enum ovsdb_jsonrpc_monitor_selection select;
 521     struct ovsdb_column_set columns;
 522 };
 523
 524 struct ovsdb_jsonrpc_monitor {
 525     struct ovsdb_replica replica;
 526     struct ovsdb_jsonrpc_session *session;
 527     struct hmap_node node;      /* In ovsdb_jsonrpc_session's "monitors". */
 528
 529     struct json *monitor_id;
 530     struct shash tables;     /* Holds "struct ovsdb_jsonrpc_monitor_table"s. */
 531 };
 532
 533 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class;
 534
 535 struct ovsdb_jsonrpc_monitor *ovsdb_jsonrpc_monitor_find(
 536     struct ovsdb_jsonrpc_session *, const struct json *monitor_id);
 537 static void ovsdb_jsonrpc_monitor_destroy(struct ovsdb_replica *);
 538 static struct json *ovsdb_jsonrpc_monitor_get_initial(
 539     const struct ovsdb_jsonrpc_monitor *);
 540
 541 static bool
 542 parse_bool(struct ovsdb_parser *parser, const char *name, bool default_value)
 543 {
 544     const struct json *json;
 545
 546     json = ovsdb_parser_member(parser, name, OP_BOOLEAN | OP_OPTIONAL);
 547     return json ? json_boolean(json) : default_value;
 548 }
 549
 550 struct ovsdb_jsonrpc_monitor *
 551 ovsdb_jsonrpc_monitor_find(struct ovsdb_jsonrpc_session *s,
 552                            const struct json *monitor_id)
 553 {
 554     struct ovsdb_jsonrpc_monitor *m;
 555
 556     HMAP_FOR_EACH_WITH_HASH (m, struct ovsdb_jsonrpc_monitor, node,
 557                              json_hash(monitor_id, 0), &s->monitors) {
 558         if (json_equal(m->monitor_id, monitor_id)) {
 559             return m;
 560         }
 561     }
 562
 563     return NULL;
 564 }
 565
 566 static struct json *
 567 ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s,
 568                              struct json *params)
 569 {
 570     struct ovsdb_jsonrpc_monitor *m = NULL;
 571     struct json *monitor_id, *monitor_requests;
 572     struct ovsdb_error *error = NULL;
 573     struct shash_node *node;
 574     struct json *json;
 575
 576     if (json_array(params)->n != 2) {
 577         error = ovsdb_syntax_error(params, NULL, "invalid parameters");
 578         goto error;
 579     }
 580     monitor_id = params->u.array.elems[0];
 581     monitor_requests = params->u.array.elems[1];
 582     if (monitor_requests->type != JSON_OBJECT) {
 583         error = ovsdb_syntax_error(monitor_requests, NULL,
 584                                    "monitor-requests must be object");
 585         goto error;
 586     }
 587
 588     if (ovsdb_jsonrpc_monitor_find(s, monitor_id)) {
 589         error = ovsdb_syntax_error(monitor_id, NULL, "duplicate monitor ID");
 590         goto error;
 591     }
 592
 593     m = xzalloc(sizeof *m);
 594     ovsdb_replica_init(&m->replica, &ovsdb_jsonrpc_replica_class);
 595     ovsdb_add_replica(s->remote->server->db, &m->replica);
 596     m->session = s;
 597     hmap_insert(&s->monitors, &m->node, json_hash(monitor_id, 0));
 598     m->monitor_id = json_clone(monitor_id);
 599     shash_init(&m->tables);
 600
 601     SHASH_FOR_EACH (node, json_object(monitor_requests)) {
 602         const struct ovsdb_table *table;
 603         struct ovsdb_jsonrpc_monitor_table *mt;
 604         const struct json *columns_json, *select_json;
 605         struct ovsdb_parser parser;
 606
 607         table = ovsdb_get_table(s->remote->server->db, node->name);
 608         if (!table) {
 609             error = ovsdb_syntax_error(NULL, NULL,
 610                                        "no table named %s", node->name);
 611             goto error;
 612         }
 613
 614         mt = xzalloc(sizeof *mt);
 615         mt->table = table;
 616         mt->select = OJMS_INITIAL | OJMS_INSERT | OJMS_DELETE | OJMS_MODIFY;
 617         ovsdb_column_set_init(&mt->columns);
 618         shash_add(&m->tables, table->schema->name, mt);
 619
 620         ovsdb_parser_init(&parser, node->data, "table %s", node->name);
 621         columns_json = ovsdb_parser_member(&parser, "columns",
 622                                            OP_ARRAY | OP_OPTIONAL);
 623         select_json = ovsdb_parser_member(&parser, "select",
 624                                           OP_OBJECT | OP_OPTIONAL);
 625         error = ovsdb_parser_finish(&parser);
 626         if (error) {
 627             goto error;
 628         }
 629
 630         if (columns_json) {
 631             error = ovsdb_column_set_from_json(columns_json, table,
 632                                                &mt->columns);
 633             if (error) {
 634                 goto error;
 635             }
 636         } else {
 637             struct shash_node *node;
 638
 639             SHASH_FOR_EACH (node, &table->schema->columns) {
 640                 const struct ovsdb_column *column = node->data;
 641                 if (column->index != OVSDB_COL_UUID) {
 642                     ovsdb_column_set_add(&mt->columns, column);
 643                 }
 644             }
 645         }
 646
 647         if (select_json) {
 648             mt->select = 0;
 649             ovsdb_parser_init(&parser, select_json, "table %s select",
 650                               table->schema->name);
 651             if (parse_bool(&parser, "initial", true)) {
 652                 mt->select |= OJMS_INITIAL;
 653             }
 654             if (parse_bool(&parser, "insert", true)) {
 655                 mt->select |= OJMS_INSERT;
 656             }
 657             if (parse_bool(&parser, "delete", true)) {
 658                 mt->select |= OJMS_DELETE;
 659             }
 660             if (parse_bool(&parser, "modify", true)) {
 661                 mt->select |= OJMS_MODIFY;
 662             }
 663             error = ovsdb_parser_finish(&parser);
 664             if (error) {
 665                 goto error;
 666             }
 667         }
 668     }
 669
 670     return ovsdb_jsonrpc_monitor_get_initial(m);
 671
 672 error:
 673     if (m) {
 674         ovsdb_remove_replica(s->remote->server->db, &m->replica);
 675     }
 676
 677     json = ovsdb_error_to_json(error);
 678     ovsdb_error_destroy(error);
 679     return json;
 680 }
 681
 682 static struct jsonrpc_msg *
 683 ovsdb_jsonrpc_monitor_cancel(struct ovsdb_jsonrpc_session *s,
 684                              struct json_array *params,
 685                              const struct json *request_id)
 686 {
 687     if (params->n != 1) {
 688         return jsonrpc_create_error(json_string_create("invalid parameters"),
 689                                     request_id);
 690     } else {
 691         struct ovsdb_jsonrpc_monitor *m;
 692
 693         m = ovsdb_jsonrpc_monitor_find(s, params->elems[0]);
 694         if (!m) {
 695             return jsonrpc_create_error(json_string_create("unknown monitor"),
 696                                         request_id);
 697         } else {
 698             ovsdb_remove_replica(s->remote->server->db, &m->replica);
 699             return jsonrpc_create_reply(json_object_create(), request_id);
 700         }
 701     }
 702 }
 703
 704 static void
 705 ovsdb_jsonrpc_monitor_remove_all(struct ovsdb_jsonrpc_session *s)
 706 {
 707     struct ovsdb_jsonrpc_monitor *m, *next;
 708
 709     HMAP_FOR_EACH_SAFE (m, next,
 710                         struct ovsdb_jsonrpc_monitor, node, &s->monitors) {
 711         ovsdb_remove_replica(s->remote->server->db, &m->replica);
 712     }
 713 }
 714
 715 static struct ovsdb_jsonrpc_monitor *
 716 ovsdb_jsonrpc_monitor_cast(struct ovsdb_replica *replica)
 717 {
 718     assert(replica->class == &ovsdb_jsonrpc_replica_class);
 719     return CONTAINER_OF(replica, struct ovsdb_jsonrpc_monitor, replica);
 720 }
 721
 722 struct ovsdb_jsonrpc_monitor_aux {
 723     bool initial;               /* Sending initial contents of table? */
 724     const struct ovsdb_jsonrpc_monitor *monitor;
 725     struct json *json;          /* JSON for the whole transaction. */
 726
 727     /* Current table.  */
 728     struct ovsdb_jsonrpc_monitor_table *mt;
 729     struct json *table_json;    /* JSON for table's transaction. */
 730 };
 731
 732 static bool
 733 ovsdb_jsonrpc_monitor_change_cb(const struct ovsdb_row *old,
 734                                 const struct ovsdb_row *new,
 735                                 void *aux_)
 736 {
 737     struct ovsdb_jsonrpc_monitor_aux *aux = aux_;
 738     const struct ovsdb_jsonrpc_monitor *m = aux->monitor;
 739     struct ovsdb_table *table = new ? new->table : old->table;
 740     enum ovsdb_jsonrpc_monitor_selection type;
 741     struct json *old_json, *new_json;
 742     struct json *row_json;
 743     char uuid[UUID_LEN + 1];
 744     int n_changed;
 745     size_t i;
 746
 747     if (!aux->mt || table != aux->mt->table) {
 748         aux->mt = shash_find_data(&m->tables, table->schema->name);
 749         aux->table_json = NULL;
 750         if (!aux->mt) {
 751             /* We don't care about rows in this table at all.  Tell the caller
 752              * to skip it.  */
 753             return false;
 754         }
 755     }
 756
 757     type = (aux->initial ? OJMS_INITIAL
 758             : !old ? OJMS_INSERT
 759             : !new ? OJMS_DELETE
 760             : OJMS_MODIFY);
 761     if (!(aux->mt->select & type)) {
 762         /* We don't care about this type of change (but do want to be called
 763          * back for changes to other rows in the same table). */
 764         return true;
 765     }
 766
 767     old_json = new_json = NULL;
 768     n_changed = 0;
 769     for (i = 0; i < aux->mt->columns.n_columns; i++) {
 770         const struct ovsdb_column *column = aux->mt->columns.columns[i];
 771         unsigned int idx = column->index;
 772         bool changed = false;
 773
 774         if (type == OJMS_MODIFY) {
 775             changed = !ovsdb_datum_equals(&old->fields[idx],
 776                                           &new->fields[idx], &column->type);
 777             n_changed += changed;
 778         }
 779         if (changed || type == OJMS_DELETE) {
 780             if (!old_json) {
 781                 old_json = json_object_create();
 782             }
 783             json_object_put(old_json, column->name,
 784                             ovsdb_datum_to_json(&old->fields[idx],
 785                                                 &column->type));
 786         }
 787         if (type & (OJMS_INITIAL | OJMS_INSERT | OJMS_MODIFY)) {
 788             if (!new_json) {
 789                 new_json = json_object_create();
 790             }
 791             json_object_put(new_json, column->name,
 792                             ovsdb_datum_to_json(&new->fields[idx],
 793                                                 &column->type));
 794         }
 795     }
 796     if ((type == OJMS_MODIFY && !n_changed) || (!old_json && !new_json)) {
 797         /* No reportable changes. */
 798         json_destroy(old_json);
 799         json_destroy(new_json);
 800         return true;
 801     }
 802
 803     /* Create JSON object for transaction overall. */
 804     if (!aux->json) {
 805         aux->json = json_object_create();
 806     }
 807
 808     /* Create JSON object for transaction on this table. */
 809     if (!aux->table_json) {
 810         aux->table_json = json_object_create();
 811         json_object_put(aux->json, aux->mt->table->schema->name,
 812                         aux->table_json);
 813     }
 814
 815     /* Create JSON object for transaction on this row. */
 816     row_json = json_object_create();
 817     if (old_json) {
 818         json_object_put(row_json, "old", old_json);
 819     }
 820     if (new_json) {
 821         json_object_put(row_json, "new", new_json);
 822     }
 823
 824     /* Add JSON row to JSON table. */
 825     snprintf(uuid, sizeof uuid,
 826              UUID_FMT, UUID_ARGS(ovsdb_row_get_uuid(new ? new : old)));
 827     json_object_put(aux->table_json, uuid, row_json);
 828
 829     return true;
 830 }
 831
 832 static void
 833 ovsdb_jsonrpc_monitor_init_aux(struct ovsdb_jsonrpc_monitor_aux *aux,
 834                                const struct ovsdb_jsonrpc_monitor *m,
 835                                bool initial)
 836 {
 837     aux->initial = initial;
 838     aux->monitor = m;
 839     aux->json = NULL;
 840     aux->mt = NULL;
 841     aux->table_json = NULL;
 842 }
 843
 844 static struct ovsdb_error *
 845 ovsdb_jsonrpc_monitor_commit(struct ovsdb_replica *replica,
 846                              const struct ovsdb_txn *txn, bool durable UNUSED)
 847 {
 848     struct ovsdb_jsonrpc_monitor *m = ovsdb_jsonrpc_monitor_cast(replica);
 849     struct ovsdb_jsonrpc_monitor_aux aux;
 850
 851     ovsdb_jsonrpc_monitor_init_aux(&aux, m, false);
 852     ovsdb_txn_for_each_change(txn, ovsdb_jsonrpc_monitor_change_cb, &aux);
 853     if (aux.json) {
 854         struct jsonrpc_msg *msg;
 855         struct json *params;
 856
 857         params = json_array_create_2(json_clone(aux.monitor->monitor_id),
 858                                      aux.json);
 859         msg = jsonrpc_create_notify("update", params);
 860         jsonrpc_session_send(aux.monitor->session->js, msg);
 861     }
 862
 863     return NULL;
 864 }
 865
 866 static struct json *
 867 ovsdb_jsonrpc_monitor_get_initial(const struct ovsdb_jsonrpc_monitor *m)
 868 {
 869     struct ovsdb_jsonrpc_monitor_aux aux;
 870     struct shash_node *node;
 871
 872     ovsdb_jsonrpc_monitor_init_aux(&aux, m, true);
 873     SHASH_FOR_EACH (node, &m->tables) {
 874         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
 875
 876         if (mt->select & OJMS_INITIAL) {
 877             struct ovsdb_row *row;
 878
 879             HMAP_FOR_EACH (row, struct ovsdb_row, hmap_node,
 880                            &mt->table->rows) {
 881                 ovsdb_jsonrpc_monitor_change_cb(NULL, row, &aux);
 882             }
 883         }
 884     }
 885     return aux.json ? aux.json : json_object_create();
 886 }
 887
 888 static void
 889 ovsdb_jsonrpc_monitor_destroy(struct ovsdb_replica *replica)
 890 {
 891     struct ovsdb_jsonrpc_monitor *m = ovsdb_jsonrpc_monitor_cast(replica);
 892     struct shash_node *node;
 893
 894     json_destroy(m->monitor_id);
 895     SHASH_FOR_EACH (node, &m->tables) {
 896         struct ovsdb_jsonrpc_monitor_table *mt = node->data;
 897         ovsdb_column_set_destroy(&mt->columns);
 898         free(mt);
 899     }
 900     shash_destroy(&m->tables);
 901     hmap_remove(&m->session->monitors, &m->node);
 902     free(m);
 903 }
 904
 905 static const struct ovsdb_replica_class ovsdb_jsonrpc_replica_class = {
 906     ovsdb_jsonrpc_monitor_commit,
 907     ovsdb_jsonrpc_monitor_destroy
 908 };