]>
git.proxmox.com Git - mirror_ovs.git/blob - lib/jsonrpc.c
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"
26 #include "fatal-signal.h"
30 #include "poll-loop.h"
31 #include "reconnect.h"
36 VLOG_DEFINE_THIS_MODULE(jsonrpc
);
39 struct stream
*stream
;
45 struct json_parser
*parser
;
46 struct jsonrpc_msg
*received
;
49 struct list output
; /* Contains "struct ofpbuf"s. */
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
*);
59 /* This is just the same as stream_open() except that it uses the default
60 * JSONRPC ports if none is specified. */
62 jsonrpc_stream_open(const char *name
, struct stream
**streamp
)
64 return stream_open_with_default_ports(name
, JSONRPC_TCP_PORT
,
65 JSONRPC_SSL_PORT
, streamp
);
68 /* This is just the same as pstream_open() except that it uses the default
69 * JSONRPC ports if none is specified. */
71 jsonrpc_pstream_open(const char *name
, struct pstream
**pstreamp
)
73 return pstream_open_with_default_ports(name
, JSONRPC_TCP_PORT
,
74 JSONRPC_SSL_PORT
, pstreamp
);
78 jsonrpc_open(struct stream
*stream
)
82 assert(stream
!= NULL
);
84 rpc
= xzalloc(sizeof *rpc
);
85 rpc
->name
= xstrdup(stream_get_name(stream
));
87 byteq_init(&rpc
->input
);
88 list_init(&rpc
->output
);
94 jsonrpc_close(struct jsonrpc
*rpc
)
104 jsonrpc_run(struct jsonrpc
*rpc
)
110 stream_run(rpc
->stream
);
111 while (!list_is_empty(&rpc
->output
)) {
112 struct ofpbuf
*buf
= ofpbuf_from_list(rpc
->output
.next
);
115 retval
= stream_send(rpc
->stream
, buf
->data
, buf
->size
);
117 rpc
->backlog
-= retval
;
118 ofpbuf_pull(buf
, retval
);
120 list_remove(&buf
->list_node
);
124 if (retval
!= -EAGAIN
) {
125 VLOG_WARN_RL(&rl
, "%s: send error: %s",
126 rpc
->name
, strerror(-retval
));
127 jsonrpc_error(rpc
, -retval
);
135 jsonrpc_wait(struct jsonrpc
*rpc
)
138 stream_run_wait(rpc
->stream
);
139 if (!list_is_empty(&rpc
->output
)) {
140 stream_send_wait(rpc
->stream
);
146 * Possible status values:
149 * - EOF: end of file (remote end closed connection; not necessarily an error)
152 jsonrpc_get_status(const struct jsonrpc
*rpc
)
158 jsonrpc_get_backlog(const struct jsonrpc
*rpc
)
160 return rpc
->status
? 0 : rpc
->backlog
;
164 jsonrpc_get_name(const struct jsonrpc
*rpc
)
170 jsonrpc_log_msg(const struct jsonrpc
*rpc
, const char *title
,
171 const struct jsonrpc_msg
*msg
)
173 if (VLOG_IS_DBG_ENABLED()) {
174 struct ds s
= DS_EMPTY_INITIALIZER
;
176 ds_put_format(&s
, ", method=\"%s\"", msg
->method
);
179 ds_put_cstr(&s
, ", params=");
180 json_to_ds(msg
->params
, 0, &s
);
183 ds_put_cstr(&s
, ", result=");
184 json_to_ds(msg
->result
, 0, &s
);
187 ds_put_cstr(&s
, ", error=");
188 json_to_ds(msg
->error
, 0, &s
);
191 ds_put_cstr(&s
, ", id=");
192 json_to_ds(msg
->id
, 0, &s
);
194 VLOG_DBG("%s: %s %s%s", rpc
->name
, title
,
195 jsonrpc_msg_type_to_string(msg
->type
), ds_cstr(&s
));
200 /* Always takes ownership of 'msg', regardless of success. */
202 jsonrpc_send(struct jsonrpc
*rpc
, struct jsonrpc_msg
*msg
)
210 jsonrpc_msg_destroy(msg
);
214 jsonrpc_log_msg(rpc
, "send", msg
);
216 json
= jsonrpc_msg_to_json(msg
);
217 s
= json_to_string(json
, 0);
221 buf
= xmalloc(sizeof *buf
);
222 ofpbuf_use(buf
, s
, length
);
224 list_push_back(&rpc
->output
, &buf
->list_node
);
225 rpc
->backlog
+= length
;
227 if (rpc
->backlog
== length
) {
234 jsonrpc_recv(struct jsonrpc
*rpc
, struct jsonrpc_msg
**msgp
)
241 while (!rpc
->received
) {
242 if (byteq_is_empty(&rpc
->input
)) {
246 chunk
= byteq_headroom(&rpc
->input
);
247 retval
= stream_recv(rpc
->stream
, byteq_head(&rpc
->input
), chunk
);
249 if (retval
== -EAGAIN
) {
252 VLOG_WARN_RL(&rl
, "%s: receive error: %s",
253 rpc
->name
, strerror(-retval
));
254 jsonrpc_error(rpc
, -retval
);
257 } else if (retval
== 0) {
258 jsonrpc_error(rpc
, EOF
);
261 byteq_advance_head(&rpc
->input
, retval
);
266 rpc
->parser
= json_parser_create(0);
268 n
= byteq_tailroom(&rpc
->input
);
269 used
= json_parser_feed(rpc
->parser
,
270 (char *) byteq_tail(&rpc
->input
), n
);
271 byteq_advance_tail(&rpc
->input
, used
);
272 if (json_parser_is_done(rpc
->parser
)) {
273 jsonrpc_received(rpc
);
275 const struct byteq
*q
= &rpc
->input
;
276 if (q
->head
<= BYTEQ_SIZE
) {
277 stream_report_content(q
->buffer
, q
->head
,
279 THIS_MODULE
, rpc
->name
);
287 *msgp
= rpc
->received
;
288 rpc
->received
= NULL
;
293 jsonrpc_recv_wait(struct jsonrpc
*rpc
)
295 if (rpc
->status
|| rpc
->received
|| !byteq_is_empty(&rpc
->input
)) {
296 poll_immediate_wake();
298 stream_recv_wait(rpc
->stream
);
302 /* Always takes ownership of 'msg', regardless of success. */
304 jsonrpc_send_block(struct jsonrpc
*rpc
, struct jsonrpc_msg
*msg
)
310 error
= jsonrpc_send(rpc
, msg
);
317 if (list_is_empty(&rpc
->output
) || rpc
->status
) {
326 jsonrpc_recv_block(struct jsonrpc
*rpc
, struct jsonrpc_msg
**msgp
)
329 int error
= jsonrpc_recv(rpc
, msgp
);
330 if (error
!= EAGAIN
) {
337 jsonrpc_recv_wait(rpc
);
342 /* Always takes ownership of 'request', regardless of success. */
344 jsonrpc_transact_block(struct jsonrpc
*rpc
, struct jsonrpc_msg
*request
,
345 struct jsonrpc_msg
**replyp
)
347 struct jsonrpc_msg
*reply
= NULL
;
351 id
= json_clone(request
->id
);
352 error
= jsonrpc_send_block(rpc
, request
);
355 error
= jsonrpc_recv_block(rpc
, &reply
);
357 || (reply
->type
== JSONRPC_REPLY
358 && json_equal(id
, reply
->id
))) {
361 jsonrpc_msg_destroy(reply
);
364 *replyp
= error
? NULL
: reply
;
370 jsonrpc_received(struct jsonrpc
*rpc
)
372 struct jsonrpc_msg
*msg
;
376 json
= json_parser_finish(rpc
->parser
);
378 if (json
->type
== JSON_STRING
) {
379 VLOG_WARN_RL(&rl
, "%s: error parsing stream: %s",
380 rpc
->name
, json_string(json
));
381 jsonrpc_error(rpc
, EPROTO
);
386 error
= jsonrpc_msg_from_json(json
, &msg
);
388 VLOG_WARN_RL(&rl
, "%s: received bad JSON-RPC message: %s",
391 jsonrpc_error(rpc
, EPROTO
);
395 jsonrpc_log_msg(rpc
, "received", msg
);
400 jsonrpc_error(struct jsonrpc
*rpc
, int error
)
405 jsonrpc_cleanup(rpc
);
410 jsonrpc_cleanup(struct jsonrpc
*rpc
)
412 stream_close(rpc
->stream
);
415 json_parser_abort(rpc
->parser
);
418 jsonrpc_msg_destroy(rpc
->received
);
419 rpc
->received
= NULL
;
421 ofpbuf_list_delete(&rpc
->output
);
425 static struct jsonrpc_msg
*
426 jsonrpc_create(enum jsonrpc_msg_type type
, const char *method
,
427 struct json
*params
, struct json
*result
, struct json
*error
,
430 struct jsonrpc_msg
*msg
= xmalloc(sizeof *msg
);
432 msg
->method
= method
? xstrdup(method
) : NULL
;
433 msg
->params
= params
;
434 msg
->result
= result
;
441 jsonrpc_create_id(void)
443 static unsigned int id
;
444 return json_integer_create(id
++);
448 jsonrpc_create_request(const char *method
, struct json
*params
,
451 struct json
*id
= jsonrpc_create_id();
453 *idp
= json_clone(id
);
455 return jsonrpc_create(JSONRPC_REQUEST
, method
, params
, NULL
, NULL
, id
);
459 jsonrpc_create_notify(const char *method
, struct json
*params
)
461 return jsonrpc_create(JSONRPC_NOTIFY
, method
, params
, NULL
, NULL
, NULL
);
465 jsonrpc_create_reply(struct json
*result
, const struct json
*id
)
467 return jsonrpc_create(JSONRPC_REPLY
, NULL
, NULL
, result
, NULL
,
472 jsonrpc_create_error(struct json
*error
, const struct json
*id
)
474 return jsonrpc_create(JSONRPC_REPLY
, NULL
, NULL
, NULL
, error
,
479 jsonrpc_msg_type_to_string(enum jsonrpc_msg_type type
)
482 case JSONRPC_REQUEST
:
486 return "notification";
498 jsonrpc_msg_is_valid(const struct jsonrpc_msg
*m
)
500 const char *type_name
;
501 unsigned int pattern
;
503 if (m
->params
&& m
->params
->type
!= JSON_ARRAY
) {
504 return xstrdup("\"params\" must be JSON array");
508 case JSONRPC_REQUEST
:
525 return xasprintf("invalid JSON-RPC message type %d", m
->type
);
528 type_name
= jsonrpc_msg_type_to_string(m
->type
);
529 if ((m
->method
!= NULL
) != ((pattern
& 0x10000) != 0)) {
530 return xasprintf("%s must%s have \"method\"",
531 type_name
, (pattern
& 0x10000) ? "" : " not");
534 if ((m
->params
!= NULL
) != ((pattern
& 0x1000) != 0)) {
535 return xasprintf("%s must%s have \"params\"",
536 type_name
, (pattern
& 0x1000) ? "" : " not");
539 if ((m
->result
!= NULL
) != ((pattern
& 0x100) != 0)) {
540 return xasprintf("%s must%s have \"result\"",
541 type_name
, (pattern
& 0x100) ? "" : " not");
544 if ((m
->error
!= NULL
) != ((pattern
& 0x10) != 0)) {
545 return xasprintf("%s must%s have \"error\"",
546 type_name
, (pattern
& 0x10) ? "" : " not");
549 if ((m
->id
!= NULL
) != ((pattern
& 0x1) != 0)) {
550 return xasprintf("%s must%s have \"id\"",
551 type_name
, (pattern
& 0x1) ? "" : " not");
558 jsonrpc_msg_destroy(struct jsonrpc_msg
*m
)
562 json_destroy(m
->params
);
563 json_destroy(m
->result
);
564 json_destroy(m
->error
);
571 null_from_json_null(struct json
*json
)
573 if (json
&& json
->type
== JSON_NULL
) {
581 jsonrpc_msg_from_json(struct json
*json
, struct jsonrpc_msg
**msgp
)
583 struct json
*method
= NULL
;
584 struct jsonrpc_msg
*msg
= NULL
;
585 struct shash
*object
;
588 if (json
->type
!= JSON_OBJECT
) {
589 error
= xstrdup("message is not a JSON object");
592 object
= json_object(json
);
594 method
= shash_find_and_delete(object
, "method");
595 if (method
&& method
->type
!= JSON_STRING
) {
596 error
= xstrdup("method is not a JSON string");
600 msg
= xzalloc(sizeof *msg
);
601 msg
->method
= method
? xstrdup(method
->u
.string
) : NULL
;
602 msg
->params
= null_from_json_null(shash_find_and_delete(object
, "params"));
603 msg
->result
= null_from_json_null(shash_find_and_delete(object
, "result"));
604 msg
->error
= null_from_json_null(shash_find_and_delete(object
, "error"));
605 msg
->id
= null_from_json_null(shash_find_and_delete(object
, "id"));
606 msg
->type
= (msg
->result
? JSONRPC_REPLY
607 : msg
->error
? JSONRPC_ERROR
608 : msg
->id
? JSONRPC_REQUEST
610 if (!shash_is_empty(object
)) {
611 error
= xasprintf("message has unexpected member \"%s\"",
612 shash_first(object
)->name
);
615 error
= jsonrpc_msg_is_valid(msg
);
621 json_destroy(method
);
624 jsonrpc_msg_destroy(msg
);
632 jsonrpc_msg_to_json(struct jsonrpc_msg
*m
)
634 struct json
*json
= json_object_create();
637 json_object_put(json
, "method", json_string_create_nocopy(m
->method
));
641 json_object_put(json
, "params", m
->params
);
645 json_object_put(json
, "result", m
->result
);
646 } else if (m
->type
== JSONRPC_ERROR
) {
647 json_object_put(json
, "result", json_null_create());
651 json_object_put(json
, "error", m
->error
);
652 } else if (m
->type
== JSONRPC_REPLY
) {
653 json_object_put(json
, "error", json_null_create());
657 json_object_put(json
, "id", m
->id
);
658 } else if (m
->type
== JSONRPC_NOTIFY
) {
659 json_object_put(json
, "id", json_null_create());
667 /* A JSON-RPC session with reconnection. */
669 struct jsonrpc_session
{
670 struct reconnect
*reconnect
;
672 struct stream
*stream
;
673 struct pstream
*pstream
;
677 /* Creates and returns a jsonrpc_session to 'name', which should be a string
678 * acceptable to stream_open() or pstream_open().
680 * If 'name' is an active connection method, e.g. "tcp:127.1.2.3", the new
681 * jsonrpc_session connects and reconnects, with back-off, to 'name'.
683 * If 'name' is a passive connection method, e.g. "ptcp:", the new
684 * jsonrpc_session listens for connections to 'name'. It maintains at most one
685 * connection at any given time. Any new connection causes the previous one
686 * (if any) to be dropped. */
687 struct jsonrpc_session
*
688 jsonrpc_session_open(const char *name
)
690 struct jsonrpc_session
*s
;
692 s
= xmalloc(sizeof *s
);
693 s
->reconnect
= reconnect_create(time_msec());
694 reconnect_set_name(s
->reconnect
, name
);
695 reconnect_enable(s
->reconnect
, time_msec());
701 if (!pstream_verify_name(name
)) {
702 reconnect_set_passive(s
->reconnect
, true, time_msec());
708 /* Creates and returns a jsonrpc_session that is initially connected to
709 * 'jsonrpc'. If the connection is dropped, it will not be reconnected.
711 * On the assumption that such connections are likely to be short-lived
712 * (e.g. from ovs-vsctl), informational logging for them is suppressed. */
713 struct jsonrpc_session
*
714 jsonrpc_session_open_unreliably(struct jsonrpc
*jsonrpc
)
716 struct jsonrpc_session
*s
;
718 s
= xmalloc(sizeof *s
);
719 s
->reconnect
= reconnect_create(time_msec());
720 reconnect_set_quiet(s
->reconnect
, true);
721 reconnect_set_name(s
->reconnect
, jsonrpc_get_name(jsonrpc
));
722 reconnect_set_max_tries(s
->reconnect
, 0);
723 reconnect_connected(s
->reconnect
, time_msec());
733 jsonrpc_session_close(struct jsonrpc_session
*s
)
736 jsonrpc_close(s
->rpc
);
737 reconnect_destroy(s
->reconnect
);
738 stream_close(s
->stream
);
739 pstream_close(s
->pstream
);
745 jsonrpc_session_disconnect(struct jsonrpc_session
*s
)
748 jsonrpc_error(s
->rpc
, EOF
);
749 jsonrpc_close(s
->rpc
);
752 } else if (s
->stream
) {
753 stream_close(s
->stream
);
760 jsonrpc_session_connect(struct jsonrpc_session
*s
)
762 const char *name
= reconnect_get_name(s
->reconnect
);
765 jsonrpc_session_disconnect(s
);
766 if (!reconnect_is_passive(s
->reconnect
)) {
767 error
= jsonrpc_stream_open(name
, &s
->stream
);
769 reconnect_connecting(s
->reconnect
, time_msec());
772 error
= s
->pstream
? 0 : jsonrpc_pstream_open(name
, &s
->pstream
);
774 reconnect_listening(s
->reconnect
, time_msec());
779 reconnect_connect_failed(s
->reconnect
, time_msec(), error
);
785 jsonrpc_session_run(struct jsonrpc_session
*s
)
788 struct stream
*stream
;
791 error
= pstream_accept(s
->pstream
, &stream
);
793 if (s
->rpc
|| s
->stream
) {
795 "%s: new connection replacing active connection",
796 reconnect_get_name(s
->reconnect
));
797 jsonrpc_session_disconnect(s
);
799 reconnect_connected(s
->reconnect
, time_msec());
800 s
->rpc
= jsonrpc_open(stream
);
801 } else if (error
!= EAGAIN
) {
802 reconnect_listen_error(s
->reconnect
, time_msec(), error
);
803 pstream_close(s
->pstream
);
812 error
= jsonrpc_get_status(s
->rpc
);
814 reconnect_disconnected(s
->reconnect
, time_msec(), error
);
815 jsonrpc_session_disconnect(s
);
817 } else if (s
->stream
) {
820 stream_run(s
->stream
);
821 error
= stream_connect(s
->stream
);
823 reconnect_connected(s
->reconnect
, time_msec());
824 s
->rpc
= jsonrpc_open(s
->stream
);
826 } else if (error
!= EAGAIN
) {
827 reconnect_connect_failed(s
->reconnect
, time_msec(), error
);
828 stream_close(s
->stream
);
833 switch (reconnect_run(s
->reconnect
, time_msec())) {
834 case RECONNECT_CONNECT
:
835 jsonrpc_session_connect(s
);
838 case RECONNECT_DISCONNECT
:
839 reconnect_disconnected(s
->reconnect
, time_msec(), 0);
840 jsonrpc_session_disconnect(s
);
843 case RECONNECT_PROBE
:
846 struct jsonrpc_msg
*request
;
848 params
= json_array_create_empty();
849 request
= jsonrpc_create_request("echo", params
, NULL
);
850 json_destroy(request
->id
);
851 request
->id
= json_string_create("echo");
852 jsonrpc_send(s
->rpc
, request
);
859 jsonrpc_session_wait(struct jsonrpc_session
*s
)
862 jsonrpc_wait(s
->rpc
);
863 } else if (s
->stream
) {
864 stream_run_wait(s
->stream
);
865 stream_connect_wait(s
->stream
);
868 pstream_wait(s
->pstream
);
870 reconnect_wait(s
->reconnect
, time_msec());
874 jsonrpc_session_get_backlog(const struct jsonrpc_session
*s
)
876 return s
->rpc
? jsonrpc_get_backlog(s
->rpc
) : 0;
880 jsonrpc_session_get_name(const struct jsonrpc_session
*s
)
882 return reconnect_get_name(s
->reconnect
);
885 /* Always takes ownership of 'msg', regardless of success. */
887 jsonrpc_session_send(struct jsonrpc_session
*s
, struct jsonrpc_msg
*msg
)
890 return jsonrpc_send(s
->rpc
, msg
);
892 jsonrpc_msg_destroy(msg
);
898 jsonrpc_session_recv(struct jsonrpc_session
*s
)
901 struct jsonrpc_msg
*msg
;
902 jsonrpc_recv(s
->rpc
, &msg
);
904 reconnect_received(s
->reconnect
, time_msec());
905 if (msg
->type
== JSONRPC_REQUEST
&& !strcmp(msg
->method
, "echo")) {
906 /* Echo request. Send reply. */
907 struct jsonrpc_msg
*reply
;
909 reply
= jsonrpc_create_reply(json_clone(msg
->params
), msg
->id
);
910 jsonrpc_session_send(s
, reply
);
911 } else if (msg
->type
== JSONRPC_REPLY
912 && msg
->id
&& msg
->id
->type
== JSON_STRING
913 && !strcmp(msg
->id
->u
.string
, "echo")) {
914 /* It's a reply to our echo request. Suppress it. */
918 jsonrpc_msg_destroy(msg
);
925 jsonrpc_session_recv_wait(struct jsonrpc_session
*s
)
928 jsonrpc_recv_wait(s
->rpc
);
933 jsonrpc_session_is_alive(const struct jsonrpc_session
*s
)
935 return s
->rpc
|| s
->stream
|| reconnect_get_max_tries(s
->reconnect
);
939 jsonrpc_session_is_connected(const struct jsonrpc_session
*s
)
941 return s
->rpc
!= NULL
;
945 jsonrpc_session_get_seqno(const struct jsonrpc_session
*s
)
951 jsonrpc_session_force_reconnect(struct jsonrpc_session
*s
)
953 reconnect_force_reconnect(s
->reconnect
, time_msec());
957 jsonrpc_session_set_max_backoff(struct jsonrpc_session
*s
, int max_backoff
)
959 reconnect_set_backoff(s
->reconnect
, 0, max_backoff
);
963 jsonrpc_session_set_probe_interval(struct jsonrpc_session
*s
,
966 reconnect_set_probe_interval(s
->reconnect
, probe_interval
);