2 * Copyright (c) 2009, 2010 Nicira Networks.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at:
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
25 #include "dynamic-string.h"
29 #include "poll-loop.h"
31 #include "reconnect.h"
35 #define THIS_MODULE VLM_jsonrpc
39 struct stream *stream;
45 struct json_parser *parser;
46 struct jsonrpc_msg *received;
49 struct ovs_queue output;
53 /* Rate limit for error messages. */
54 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
56 static void jsonrpc_received(struct jsonrpc *);
57 static void jsonrpc_cleanup(struct jsonrpc *);
60 jsonrpc_open(struct stream *stream)
64 assert(stream != NULL);
66 rpc = xzalloc(sizeof *rpc);
67 rpc->name = xstrdup(stream_get_name(stream));
69 byteq_init(&rpc->input);
70 queue_init(&rpc->output);
76 jsonrpc_close(struct jsonrpc *rpc)
86 jsonrpc_run(struct jsonrpc *rpc)
92 stream_run(rpc->stream);
93 while (!queue_is_empty(&rpc->output)) {
94 struct ofpbuf *buf = rpc->output.head;
97 retval = stream_send(rpc->stream, buf->data, buf->size);
99 rpc->backlog -= retval;
100 ofpbuf_pull(buf, retval);
102 ofpbuf_delete(queue_pop_head(&rpc->output));
105 if (retval != -EAGAIN) {
106 VLOG_WARN_RL(&rl, "%s: send error: %s",
107 rpc->name, strerror(-retval));
108 jsonrpc_error(rpc, -retval);
116 jsonrpc_wait(struct jsonrpc *rpc)
119 stream_run_wait(rpc->stream);
120 if (!queue_is_empty(&rpc->output)) {
121 stream_send_wait(rpc->stream);
127 jsonrpc_get_status(const struct jsonrpc *rpc)
133 jsonrpc_get_backlog(const struct jsonrpc *rpc)
135 return rpc->status ? 0 : rpc->backlog;
139 jsonrpc_get_name(const struct jsonrpc *rpc)
145 jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title,
146 const struct jsonrpc_msg *msg)
148 if (VLOG_IS_DBG_ENABLED()) {
149 struct ds s = DS_EMPTY_INITIALIZER;
151 ds_put_format(&s, ", method=\"%s\"", msg->method);
154 ds_put_cstr(&s, ", params=");
155 json_to_ds(msg->params, 0, &s);
158 ds_put_cstr(&s, ", result=");
159 json_to_ds(msg->result, 0, &s);
162 ds_put_cstr(&s, ", error=");
163 json_to_ds(msg->error, 0, &s);
166 ds_put_cstr(&s, ", id=");
167 json_to_ds(msg->id, 0, &s);
169 VLOG_DBG("%s: %s %s%s", rpc->name, title,
170 jsonrpc_msg_type_to_string(msg->type), ds_cstr(&s));
175 /* Always takes ownership of 'msg', regardless of success. */
177 jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
185 jsonrpc_msg_destroy(msg);
189 jsonrpc_log_msg(rpc, "send", msg);
191 json = jsonrpc_msg_to_json(msg);
192 s = json_to_string(json, 0);
196 buf = xmalloc(sizeof *buf);
197 ofpbuf_use(buf, s, length);
199 queue_push_tail(&rpc->output, buf);
200 rpc->backlog += length;
202 if (rpc->output.n == 1) {
209 jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
216 while (!rpc->received) {
217 if (byteq_is_empty(&rpc->input)) {
221 chunk = byteq_headroom(&rpc->input);
222 retval = stream_recv(rpc->stream, byteq_head(&rpc->input), chunk);
224 if (retval == -EAGAIN) {
227 VLOG_WARN_RL(&rl, "%s: receive error: %s",
228 rpc->name, strerror(-retval));
229 jsonrpc_error(rpc, -retval);
232 } else if (retval == 0) {
233 VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name);
234 jsonrpc_error(rpc, EOF);
237 byteq_advance_head(&rpc->input, retval);
242 rpc->parser = json_parser_create(0);
244 n = byteq_tailroom(&rpc->input);
245 used = json_parser_feed(rpc->parser,
246 (char *) byteq_tail(&rpc->input), n);
247 byteq_advance_tail(&rpc->input, used);
248 if (json_parser_is_done(rpc->parser)) {
249 jsonrpc_received(rpc);
257 *msgp = rpc->received;
258 rpc->received = NULL;
263 jsonrpc_recv_wait(struct jsonrpc *rpc)
265 if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) {
266 poll_immediate_wake();
268 stream_recv_wait(rpc->stream);
272 /* Always takes ownership of 'msg', regardless of success. */
274 jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
278 error = jsonrpc_send(rpc, msg);
285 if (queue_is_empty(&rpc->output) || rpc->status) {
294 jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
297 int error = jsonrpc_recv(rpc, msgp);
298 if (error != EAGAIN) {
304 jsonrpc_recv_wait(rpc);
309 /* Always takes ownership of 'request', regardless of success. */
311 jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request,
312 struct jsonrpc_msg **replyp)
314 struct jsonrpc_msg *reply = NULL;
318 id = json_clone(request->id);
319 error = jsonrpc_send_block(rpc, request);
322 error = jsonrpc_recv_block(rpc, &reply);
324 || (reply->type == JSONRPC_REPLY
325 && json_equal(id, reply->id))) {
328 jsonrpc_msg_destroy(reply);
331 *replyp = error ? NULL : reply;
337 jsonrpc_received(struct jsonrpc *rpc)
339 struct jsonrpc_msg *msg;
343 json = json_parser_finish(rpc->parser);
345 if (json->type == JSON_STRING) {
346 VLOG_WARN_RL(&rl, "%s: error parsing stream: %s",
347 rpc->name, json_string(json));
348 jsonrpc_error(rpc, EPROTO);
353 error = jsonrpc_msg_from_json(json, &msg);
355 VLOG_WARN_RL(&rl, "%s: received bad JSON-RPC message: %s",
358 jsonrpc_error(rpc, EPROTO);
362 jsonrpc_log_msg(rpc, "received", msg);
367 jsonrpc_error(struct jsonrpc *rpc, int error)
372 jsonrpc_cleanup(rpc);
377 jsonrpc_cleanup(struct jsonrpc *rpc)
379 stream_close(rpc->stream);
382 json_parser_abort(rpc->parser);
385 jsonrpc_msg_destroy(rpc->received);
386 rpc->received = NULL;
388 queue_clear(&rpc->output);
392 static struct jsonrpc_msg *
393 jsonrpc_create(enum jsonrpc_msg_type type, const char *method,
394 struct json *params, struct json *result, struct json *error,
397 struct jsonrpc_msg *msg = xmalloc(sizeof *msg);
399 msg->method = method ? xstrdup(method) : NULL;
400 msg->params = params;
401 msg->result = result;
408 jsonrpc_create_id(void)
410 static unsigned int id;
411 return json_integer_create(id++);
415 jsonrpc_create_request(const char *method, struct json *params,
418 struct json *id = jsonrpc_create_id();
420 *idp = json_clone(id);
422 return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, id);
426 jsonrpc_create_notify(const char *method, struct json *params)
428 return jsonrpc_create(JSONRPC_NOTIFY, method, params, NULL, NULL, NULL);
432 jsonrpc_create_reply(struct json *result, const struct json *id)
434 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, result, NULL,
439 jsonrpc_create_error(struct json *error, const struct json *id)
441 return jsonrpc_create(JSONRPC_REPLY, NULL, NULL, NULL, error,
446 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type)
449 case JSONRPC_REQUEST:
453 return "notification";
465 jsonrpc_msg_is_valid(const struct jsonrpc_msg *m)
467 const char *type_name;
468 unsigned int pattern;
470 if (m->params && m->params->type != JSON_ARRAY) {
471 return xstrdup("\"params\" must be JSON array");
475 case JSONRPC_REQUEST:
492 return xasprintf("invalid JSON-RPC message type %d", m->type);
495 type_name = jsonrpc_msg_type_to_string(m->type);
496 if ((m->method != NULL) != ((pattern & 0x10000) != 0)) {
497 return xasprintf("%s must%s have \"method\"",
498 type_name, (pattern & 0x10000) ? "" : " not");
501 if ((m->params != NULL) != ((pattern & 0x1000) != 0)) {
502 return xasprintf("%s must%s have \"params\"",
503 type_name, (pattern & 0x1000) ? "" : " not");
506 if ((m->result != NULL) != ((pattern & 0x100) != 0)) {
507 return xasprintf("%s must%s have \"result\"",
508 type_name, (pattern & 0x100) ? "" : " not");
511 if ((m->error != NULL) != ((pattern & 0x10) != 0)) {
512 return xasprintf("%s must%s have \"error\"",
513 type_name, (pattern & 0x10) ? "" : " not");
516 if ((m->id != NULL) != ((pattern & 0x1) != 0)) {
517 return xasprintf("%s must%s have \"id\"",
518 type_name, (pattern & 0x1) ? "" : " not");
525 jsonrpc_msg_destroy(struct jsonrpc_msg *m)
529 json_destroy(m->params);
530 json_destroy(m->result);
531 json_destroy(m->error);
538 null_from_json_null(struct json *json)
540 if (json && json->type == JSON_NULL) {
548 jsonrpc_msg_from_json(struct json *json, struct jsonrpc_msg **msgp)
550 struct json *method = NULL;
551 struct jsonrpc_msg *msg = NULL;
552 struct shash *object;
555 if (json->type != JSON_OBJECT) {
556 error = xstrdup("message is not a JSON object");
559 object = json_object(json);
561 method = shash_find_and_delete(object, "method");
562 if (method && method->type != JSON_STRING) {
563 error = xstrdup("method is not a JSON string");
567 msg = xzalloc(sizeof *msg);
568 msg->method = method ? xstrdup(method->u.string) : NULL;
569 msg->params = null_from_json_null(shash_find_and_delete(object, "params"));
570 msg->result = null_from_json_null(shash_find_and_delete(object, "result"));
571 msg->error = null_from_json_null(shash_find_and_delete(object, "error"));
572 msg->id = null_from_json_null(shash_find_and_delete(object, "id"));
573 msg->type = (msg->result ? JSONRPC_REPLY
574 : msg->error ? JSONRPC_ERROR
575 : msg->id ? JSONRPC_REQUEST
577 if (!shash_is_empty(object)) {
578 error = xasprintf("message has unexpected member \"%s\"",
579 shash_first(object)->name);
582 error = jsonrpc_msg_is_valid(msg);
588 json_destroy(method);
591 jsonrpc_msg_destroy(msg);
599 jsonrpc_msg_to_json(struct jsonrpc_msg *m)
601 struct json *json = json_object_create();
604 json_object_put(json, "method", json_string_create_nocopy(m->method));
608 json_object_put(json, "params", m->params);
612 json_object_put(json, "result", m->result);
613 } else if (m->type == JSONRPC_ERROR) {
614 json_object_put(json, "result", json_null_create());
618 json_object_put(json, "error", m->error);
619 } else if (m->type == JSONRPC_REPLY) {
620 json_object_put(json, "error", json_null_create());
624 json_object_put(json, "id", m->id);
625 } else if (m->type == JSONRPC_NOTIFY) {
626 json_object_put(json, "id", json_null_create());
634 /* A JSON-RPC session with reconnection. */
636 struct jsonrpc_session {
637 struct reconnect *reconnect;
639 struct stream *stream;
643 /* Creates and returns a jsonrpc_session that connects and reconnects, with
644 * back-off, to 'name', which should be a string acceptable to
646 struct jsonrpc_session *
647 jsonrpc_session_open(const char *name)
649 struct jsonrpc_session *s;
651 s = xmalloc(sizeof *s);
652 s->reconnect = reconnect_create(time_msec());
653 reconnect_set_name(s->reconnect, name);
654 reconnect_enable(s->reconnect, time_msec());
662 /* Creates and returns a jsonrpc_session that is initially connected to
663 * 'jsonrpc'. If the connection is dropped, it will not be reconnected. */
664 struct jsonrpc_session *
665 jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc)
667 struct jsonrpc_session *s;
669 s = xmalloc(sizeof *s);
670 s->reconnect = reconnect_create(time_msec());
671 reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc));
672 reconnect_set_max_tries(s->reconnect, 0);
673 reconnect_connected(s->reconnect, time_msec());
682 jsonrpc_session_close(struct jsonrpc_session *s)
685 jsonrpc_close(s->rpc);
686 reconnect_destroy(s->reconnect);
692 jsonrpc_session_disconnect(struct jsonrpc_session *s)
695 jsonrpc_error(s->rpc, EOF);
696 jsonrpc_close(s->rpc);
699 } else if (s->stream) {
700 stream_close(s->stream);
707 jsonrpc_session_connect(struct jsonrpc_session *s)
711 jsonrpc_session_disconnect(s);
712 error = stream_open(reconnect_get_name(s->reconnect), &s->stream);
714 reconnect_connect_failed(s->reconnect, time_msec(), error);
716 reconnect_connecting(s->reconnect, time_msec());
722 jsonrpc_session_run(struct jsonrpc_session *s)
728 error = jsonrpc_get_status(s->rpc);
730 reconnect_disconnected(s->reconnect, time_msec(), 0);
731 jsonrpc_session_disconnect(s);
733 } else if (s->stream) {
736 stream_run(s->stream);
737 error = stream_connect(s->stream);
739 reconnect_connected(s->reconnect, time_msec());
740 s->rpc = jsonrpc_open(s->stream);
742 } else if (error != EAGAIN) {
743 reconnect_connect_failed(s->reconnect, time_msec(), error);
744 stream_close(s->stream);
749 switch (reconnect_run(s->reconnect, time_msec())) {
750 case RECONNECT_CONNECT:
751 jsonrpc_session_connect(s);
754 case RECONNECT_DISCONNECT:
755 reconnect_disconnected(s->reconnect, time_msec(), 0);
756 jsonrpc_session_disconnect(s);
759 case RECONNECT_PROBE:
762 struct jsonrpc_msg *request;
764 params = json_array_create_empty();
765 request = jsonrpc_create_request("echo", params, NULL);
766 json_destroy(request->id);
767 request->id = json_string_create("echo");
768 jsonrpc_send(s->rpc, request);
775 jsonrpc_session_wait(struct jsonrpc_session *s)
778 jsonrpc_wait(s->rpc);
779 } else if (s->stream) {
780 stream_run_wait(s->stream);
781 stream_connect_wait(s->stream);
783 reconnect_wait(s->reconnect, time_msec());
787 jsonrpc_session_get_backlog(const struct jsonrpc_session *s)
789 return s->rpc ? jsonrpc_get_backlog(s->rpc) : 0;
793 jsonrpc_session_get_name(const struct jsonrpc_session *s)
795 return reconnect_get_name(s->reconnect);
798 /* Always takes ownership of 'msg', regardless of success. */
800 jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg)
803 return jsonrpc_send(s->rpc, msg);
805 jsonrpc_msg_destroy(msg);
811 jsonrpc_session_recv(struct jsonrpc_session *s)
814 struct jsonrpc_msg *msg;
815 jsonrpc_recv(s->rpc, &msg);
817 reconnect_received(s->reconnect, time_msec());
818 if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
819 /* Echo request. Send reply. */
820 struct jsonrpc_msg *reply;
822 reply = jsonrpc_create_reply(json_clone(msg->params), msg->id);
823 jsonrpc_session_send(s, reply);
824 } else if (msg->type == JSONRPC_REPLY
825 && msg->id && msg->id->type == JSON_STRING
826 && !strcmp(msg->id->u.string, "echo")) {
827 /* It's a reply to our echo request. Suppress it. */
831 jsonrpc_msg_destroy(msg);
838 jsonrpc_session_recv_wait(struct jsonrpc_session *s)
841 jsonrpc_recv_wait(s->rpc);
846 jsonrpc_session_is_alive(const struct jsonrpc_session *s)
848 return s->rpc || s->stream || reconnect_get_max_tries(s->reconnect);
852 jsonrpc_session_is_connected(const struct jsonrpc_session *s)
854 return s->rpc != NULL;
858 jsonrpc_session_get_seqno(const struct jsonrpc_session *s)
864 jsonrpc_session_force_reconnect(struct jsonrpc_session *s)
866 reconnect_force_reconnect(s->reconnect, time_msec());