X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=09b10711ff851fd8ca861cbb1af429df14b0f842;hb=94e0c3ff3c096456bddfe33ac8d6540be17bfe8e;hp=cf2a497156334df3f5d883a8e2fe7afd31f93c71;hpb=f212909325be9bc7e296e1a32e2fc89694a0049f;p=sliver-openvswitch.git diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c index cf2a49715..09b10711f 100644 --- a/lib/jsonrpc.c +++ b/lib/jsonrpc.c @@ -1,5 +1,5 @@ /* - * Copyright (c) 2009 Nicira Networks. + * Copyright (c) 2009, 2010, 2011, 2012 Nicira Networks. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,18 +18,22 @@ #include "jsonrpc.h" +#include #include #include "byteq.h" +#include "dynamic-string.h" +#include "fatal-signal.h" #include "json.h" #include "list.h" #include "ofpbuf.h" #include "poll-loop.h" -#include "queue.h" +#include "reconnect.h" #include "stream.h" - -#define THIS_MODULE VLM_jsonrpc +#include "timeval.h" #include "vlog.h" + +VLOG_DEFINE_THIS_MODULE(jsonrpc); struct jsonrpc { struct stream *stream; @@ -42,7 +46,7 @@ struct jsonrpc { struct jsonrpc_msg *received; /* Output. */ - struct ovs_queue output; + struct list output; /* Contains "struct ofpbuf"s. */ size_t backlog; }; @@ -51,7 +55,28 @@ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5); static void jsonrpc_received(struct jsonrpc *); static void jsonrpc_cleanup(struct jsonrpc *); +static void jsonrpc_error(struct jsonrpc *, int error); + +/* This is just the same as stream_open() except that it uses the default + * JSONRPC ports if none is specified. */ +int +jsonrpc_stream_open(const char *name, struct stream **streamp) +{ + return stream_open_with_default_ports(name, JSONRPC_TCP_PORT, + JSONRPC_SSL_PORT, streamp); +} +/* This is just the same as pstream_open() except that it uses the default + * JSONRPC ports if none is specified. */ +int +jsonrpc_pstream_open(const char *name, struct pstream **pstreamp) +{ + return pstream_open_with_default_ports(name, JSONRPC_TCP_PORT, + JSONRPC_SSL_PORT, pstreamp); +} + +/* Returns a new JSON-RPC stream that uses 'stream' for input and output. The + * new jsonrpc object takes ownership of 'stream'. */ struct jsonrpc * jsonrpc_open(struct stream *stream) { @@ -63,11 +88,13 @@ jsonrpc_open(struct stream *stream) rpc->name = xstrdup(stream_get_name(stream)); rpc->stream = stream; byteq_init(&rpc->input); - queue_init(&rpc->output); + list_init(&rpc->output); return rpc; } +/* Destroys 'rpc', closing the stream on which it is based, and frees its + * memory. */ void jsonrpc_close(struct jsonrpc *rpc) { @@ -78,6 +105,7 @@ jsonrpc_close(struct jsonrpc *rpc) } } +/* Performs periodic maintenance on 'rpc', such as flushing output buffers. */ void jsonrpc_run(struct jsonrpc *rpc) { @@ -85,8 +113,9 @@ jsonrpc_run(struct jsonrpc *rpc) return; } - while (!queue_is_empty(&rpc->output)) { - struct ofpbuf *buf = rpc->output.head; + stream_run(rpc->stream); + while (!list_is_empty(&rpc->output)) { + struct ofpbuf *buf = ofpbuf_from_list(rpc->output.next); int retval; retval = stream_send(rpc->stream, buf->data, buf->size); @@ -94,7 +123,8 @@ jsonrpc_run(struct jsonrpc *rpc) rpc->backlog -= retval; ofpbuf_pull(buf, retval); if (!buf->size) { - ofpbuf_delete(queue_pop_head(&rpc->output)); + list_remove(&buf->list_node); + ofpbuf_delete(buf); } } else { if (retval != -EAGAIN) { @@ -107,32 +137,94 @@ jsonrpc_run(struct jsonrpc *rpc) } } +/* Arranges for the poll loop to wake up when 'rpc' needs to perform + * maintenance activities. */ void jsonrpc_wait(struct jsonrpc *rpc) { - if (!rpc->status && !queue_is_empty(&rpc->output)) { - stream_send_wait(rpc->stream); + if (!rpc->status) { + stream_run_wait(rpc->stream); + if (!list_is_empty(&rpc->output)) { + stream_send_wait(rpc->stream); + } } } +/* + * Returns the current status of 'rpc'. The possible return values are: + * - 0: no error yet + * - >0: errno value + * - EOF: end of file (remote end closed connection; not necessarily an error). + * + * When this functions nonzero, 'rpc' is effectively out of commission. 'rpc' + * will not receive any more messages and any further messages that one + * attempts to send with 'rpc' will be discarded. The caller can keep 'rpc' + * around as long as it wants, but it's not going to provide any more useful + * services. + */ int jsonrpc_get_status(const struct jsonrpc *rpc) { return rpc->status; } +/* Returns the number of bytes buffered by 'rpc' to be written to the + * underlying stream. Always returns 0 if 'rpc' has encountered an error or if + * the remote end closed the connection. */ size_t jsonrpc_get_backlog(const struct jsonrpc *rpc) { return rpc->status ? 0 : rpc->backlog; } +/* Returns 'rpc''s name, that is, the name returned by stream_get_name() for + * the stream underlying 'rpc' when 'rpc' was created. */ const char * jsonrpc_get_name(const struct jsonrpc *rpc) { return rpc->name; } +static void +jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title, + const struct jsonrpc_msg *msg) +{ + if (VLOG_IS_DBG_ENABLED()) { + struct ds s = DS_EMPTY_INITIALIZER; + if (msg->method) { + ds_put_format(&s, ", method=\"%s\"", msg->method); + } + if (msg->params) { + ds_put_cstr(&s, ", params="); + json_to_ds(msg->params, 0, &s); + } + if (msg->result) { + ds_put_cstr(&s, ", result="); + json_to_ds(msg->result, 0, &s); + } + if (msg->error) { + ds_put_cstr(&s, ", error="); + json_to_ds(msg->error, 0, &s); + } + if (msg->id) { + ds_put_cstr(&s, ", id="); + json_to_ds(msg->id, 0, &s); + } + VLOG_DBG("%s: %s %s%s", rpc->name, title, + jsonrpc_msg_type_to_string(msg->type), ds_cstr(&s)); + ds_destroy(&s); + } +} + +/* Schedules 'msg' to be sent on 'rpc' and returns 'rpc''s status (as with + * jsonrpc_get_status()). + * + * If 'msg' cannot be sent immediately, it is appended to a buffer. The caller + * is responsible for ensuring that the amount of buffered data is somehow + * limited. (jsonrpc_get_backlog() returns the amount of data currently + * buffered in 'rpc'.) + * + * Always takes ownership of 'msg', regardless of success. */ int jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg) { @@ -146,6 +238,8 @@ jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg) return rpc->status; } + jsonrpc_log_msg(rpc, "send", msg); + json = jsonrpc_msg_to_json(msg); s = json_to_string(json, 0); length = strlen(s); @@ -154,15 +248,31 @@ jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg) buf = xmalloc(sizeof *buf); ofpbuf_use(buf, s, length); buf->size = length; - queue_push_tail(&rpc->output, buf); + list_push_back(&rpc->output, &buf->list_node); rpc->backlog += length; - if (rpc->output.n == 1) { + if (rpc->backlog == length) { jsonrpc_run(rpc); } return rpc->status; } +/* Attempts to receive a message from 'rpc'. + * + * If successful, stores the received message in '*msgp' and returns 0. The + * caller takes ownership of '*msgp' and must eventually destroy it with + * jsonrpc_msg_destroy(). + * + * Otherwise, stores NULL in '*msgp' and returns one of the following: + * + * - EAGAIN: No message has been received. + * + * - EOF: The remote end closed the connection gracefully. + * + * - Otherwise an errno value that represents a JSON-RPC protocol violation + * or another error fatal to the connection. 'rpc' will not send or + * receive any more messages. + */ int jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) { @@ -188,7 +298,6 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) return rpc->status; } } else if (retval == 0) { - VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name); jsonrpc_error(rpc, EOF); return EOF; } @@ -206,6 +315,12 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) if (json_parser_is_done(rpc->parser)) { jsonrpc_received(rpc); if (rpc->status) { + const struct byteq *q = &rpc->input; + if (q->head <= BYTEQ_SIZE) { + stream_report_content(q->buffer, q->head, + STREAM_JSONRPC, + THIS_MODULE, rpc->name); + } return rpc->status; } } @@ -217,40 +332,54 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) return 0; } +/* Causes the poll loop to wake up when jsonrpc_recv() may return a value other + * than EAGAIN. */ void jsonrpc_recv_wait(struct jsonrpc *rpc) { if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) { - poll_immediate_wake(); + (poll_immediate_wake)(rpc->name); } else { stream_recv_wait(rpc->stream); } } +/* Sends 'msg' on 'rpc' and waits for it to be successfully queued to the + * underlying stream. Returns 0 if 'msg' was sent successfully, otherwise a + * status value (see jsonrpc_get_status()). + * + * Always takes ownership of 'msg', regardless of success. */ int jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg) { int error; + fatal_signal_run(); + error = jsonrpc_send(rpc, msg); if (error) { return error; } - while (!queue_is_empty(&rpc->output) && !rpc->status) { + for (;;) { jsonrpc_run(rpc); + if (list_is_empty(&rpc->output) || rpc->status) { + return rpc->status; + } jsonrpc_wait(rpc); poll_block(); } - return rpc->status; } +/* Waits for a message to be received on 'rpc'. Same semantics as + * jsonrpc_recv() except that EAGAIN will never be returned. */ int jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) { for (;;) { int error = jsonrpc_recv(rpc, msgp); if (error != EAGAIN) { + fatal_signal_run(); return error; } @@ -261,6 +390,43 @@ jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) } } +/* Sends 'request' to 'rpc' then waits for a reply. The return value is 0 if + * successful, in which case '*replyp' is set to the reply, which the caller + * must eventually free with jsonrpc_msg_destroy(). Otherwise returns a status + * value (see jsonrpc_get_status()). + * + * Discards any message received on 'rpc' that is not a reply to 'request' + * (based on message id). + * + * Always takes ownership of 'request', regardless of success. */ +int +jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request, + struct jsonrpc_msg **replyp) +{ + struct jsonrpc_msg *reply = NULL; + struct json *id; + int error; + + id = json_clone(request->id); + error = jsonrpc_send_block(rpc, request); + if (!error) { + for (;;) { + error = jsonrpc_recv_block(rpc, &reply); + if (error) { + break; + } + if ((reply->type == JSONRPC_REPLY || reply->type == JSONRPC_ERROR) + && json_equal(id, reply->id)) { + break; + } + jsonrpc_msg_destroy(reply); + } + } + *replyp = error ? NULL : reply; + json_destroy(id); + return error; +} + static void jsonrpc_received(struct jsonrpc *rpc) { @@ -287,10 +453,11 @@ jsonrpc_received(struct jsonrpc *rpc) return; } + jsonrpc_log_msg(rpc, "received", msg); rpc->received = msg; } -void +static void jsonrpc_error(struct jsonrpc *rpc, int error) { assert(error); @@ -312,7 +479,7 @@ jsonrpc_cleanup(struct jsonrpc *rpc) jsonrpc_msg_destroy(rpc->received); rpc->received = NULL; - queue_clear(&rpc->output); + ofpbuf_list_delete(&rpc->output); rpc->backlog = 0; } @@ -339,10 +506,14 @@ jsonrpc_create_id(void) } struct jsonrpc_msg * -jsonrpc_create_request(const char *method, struct json *params) +jsonrpc_create_request(const char *method, struct json *params, + struct json **idp) { - return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, - jsonrpc_create_id()); + struct json *id = jsonrpc_create_id(); + if (idp) { + *idp = json_clone(id); + } + return jsonrpc_create(JSONRPC_REQUEST, method, params, NULL, NULL, id); } struct jsonrpc_msg * @@ -553,3 +724,320 @@ jsonrpc_msg_to_json(struct jsonrpc_msg *m) return json; } + +/* A JSON-RPC session with reconnection. */ + +struct jsonrpc_session { + struct reconnect *reconnect; + struct jsonrpc *rpc; + struct stream *stream; + struct pstream *pstream; + unsigned int seqno; +}; + +/* Creates and returns a jsonrpc_session to 'name', which should be a string + * acceptable to stream_open() or pstream_open(). + * + * If 'name' is an active connection method, e.g. "tcp:127.1.2.3", the new + * jsonrpc_session connects and reconnects, with back-off, to 'name'. + * + * If 'name' is a passive connection method, e.g. "ptcp:", the new + * jsonrpc_session listens for connections to 'name'. It maintains at most one + * connection at any given time. Any new connection causes the previous one + * (if any) to be dropped. */ +struct jsonrpc_session * +jsonrpc_session_open(const char *name) +{ + struct jsonrpc_session *s; + + s = xmalloc(sizeof *s); + s->reconnect = reconnect_create(time_msec()); + reconnect_set_name(s->reconnect, name); + reconnect_enable(s->reconnect, time_msec()); + s->rpc = NULL; + s->stream = NULL; + s->pstream = NULL; + s->seqno = 0; + + if (!pstream_verify_name(name)) { + reconnect_set_passive(s->reconnect, true, time_msec()); + } + + return s; +} + +/* Creates and returns a jsonrpc_session that is initially connected to + * 'jsonrpc'. If the connection is dropped, it will not be reconnected. + * + * On the assumption that such connections are likely to be short-lived + * (e.g. from ovs-vsctl), informational logging for them is suppressed. */ +struct jsonrpc_session * +jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc) +{ + struct jsonrpc_session *s; + + s = xmalloc(sizeof *s); + s->reconnect = reconnect_create(time_msec()); + reconnect_set_quiet(s->reconnect, true); + reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc)); + reconnect_set_max_tries(s->reconnect, 0); + reconnect_connected(s->reconnect, time_msec()); + s->rpc = jsonrpc; + s->stream = NULL; + s->pstream = NULL; + s->seqno = 0; + + return s; +} + +void +jsonrpc_session_close(struct jsonrpc_session *s) +{ + if (s) { + jsonrpc_close(s->rpc); + reconnect_destroy(s->reconnect); + stream_close(s->stream); + pstream_close(s->pstream); + free(s); + } +} + +static void +jsonrpc_session_disconnect(struct jsonrpc_session *s) +{ + if (s->rpc) { + jsonrpc_error(s->rpc, EOF); + jsonrpc_close(s->rpc); + s->rpc = NULL; + s->seqno++; + } else if (s->stream) { + stream_close(s->stream); + s->stream = NULL; + s->seqno++; + } +} + +static void +jsonrpc_session_connect(struct jsonrpc_session *s) +{ + const char *name = reconnect_get_name(s->reconnect); + int error; + + jsonrpc_session_disconnect(s); + if (!reconnect_is_passive(s->reconnect)) { + error = jsonrpc_stream_open(name, &s->stream); + if (!error) { + reconnect_connecting(s->reconnect, time_msec()); + } + } else { + error = s->pstream ? 0 : jsonrpc_pstream_open(name, &s->pstream); + if (!error) { + reconnect_listening(s->reconnect, time_msec()); + } + } + + if (error) { + reconnect_connect_failed(s->reconnect, time_msec(), error); + } + s->seqno++; +} + +void +jsonrpc_session_run(struct jsonrpc_session *s) +{ + if (s->pstream) { + struct stream *stream; + int error; + + error = pstream_accept(s->pstream, &stream); + if (!error) { + if (s->rpc || s->stream) { + VLOG_INFO_RL(&rl, + "%s: new connection replacing active connection", + reconnect_get_name(s->reconnect)); + jsonrpc_session_disconnect(s); + } + reconnect_connected(s->reconnect, time_msec()); + s->rpc = jsonrpc_open(stream); + } else if (error != EAGAIN) { + reconnect_listen_error(s->reconnect, time_msec(), error); + pstream_close(s->pstream); + s->pstream = NULL; + } + } + + if (s->rpc) { + int error; + + jsonrpc_run(s->rpc); + error = jsonrpc_get_status(s->rpc); + if (error) { + reconnect_disconnected(s->reconnect, time_msec(), error); + jsonrpc_session_disconnect(s); + } + } else if (s->stream) { + int error; + + stream_run(s->stream); + error = stream_connect(s->stream); + if (!error) { + reconnect_connected(s->reconnect, time_msec()); + s->rpc = jsonrpc_open(s->stream); + s->stream = NULL; + } else if (error != EAGAIN) { + reconnect_connect_failed(s->reconnect, time_msec(), error); + stream_close(s->stream); + s->stream = NULL; + } + } + + switch (reconnect_run(s->reconnect, time_msec())) { + case RECONNECT_CONNECT: + jsonrpc_session_connect(s); + break; + + case RECONNECT_DISCONNECT: + reconnect_disconnected(s->reconnect, time_msec(), 0); + jsonrpc_session_disconnect(s); + break; + + case RECONNECT_PROBE: + if (s->rpc) { + struct json *params; + struct jsonrpc_msg *request; + + params = json_array_create_empty(); + request = jsonrpc_create_request("echo", params, NULL); + json_destroy(request->id); + request->id = json_string_create("echo"); + jsonrpc_send(s->rpc, request); + } + break; + } +} + +void +jsonrpc_session_wait(struct jsonrpc_session *s) +{ + if (s->rpc) { + jsonrpc_wait(s->rpc); + } else if (s->stream) { + stream_run_wait(s->stream); + stream_connect_wait(s->stream); + } + if (s->pstream) { + pstream_wait(s->pstream); + } + reconnect_wait(s->reconnect, time_msec()); +} + +size_t +jsonrpc_session_get_backlog(const struct jsonrpc_session *s) +{ + return s->rpc ? jsonrpc_get_backlog(s->rpc) : 0; +} + +/* Always returns a pointer to a valid C string, assuming 's' was initialized + * correctly. */ +const char * +jsonrpc_session_get_name(const struct jsonrpc_session *s) +{ + return reconnect_get_name(s->reconnect); +} + +/* Always takes ownership of 'msg', regardless of success. */ +int +jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg) +{ + if (s->rpc) { + return jsonrpc_send(s->rpc, msg); + } else { + jsonrpc_msg_destroy(msg); + return ENOTCONN; + } +} + +struct jsonrpc_msg * +jsonrpc_session_recv(struct jsonrpc_session *s) +{ + if (s->rpc) { + struct jsonrpc_msg *msg; + jsonrpc_recv(s->rpc, &msg); + if (msg) { + reconnect_received(s->reconnect, time_msec()); + if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) { + /* Echo request. Send reply. */ + struct jsonrpc_msg *reply; + + reply = jsonrpc_create_reply(json_clone(msg->params), msg->id); + jsonrpc_session_send(s, reply); + } else if (msg->type == JSONRPC_REPLY + && msg->id && msg->id->type == JSON_STRING + && !strcmp(msg->id->u.string, "echo")) { + /* It's a reply to our echo request. Suppress it. */ + } else { + return msg; + } + jsonrpc_msg_destroy(msg); + } + } + return NULL; +} + +void +jsonrpc_session_recv_wait(struct jsonrpc_session *s) +{ + if (s->rpc) { + jsonrpc_recv_wait(s->rpc); + } +} + +bool +jsonrpc_session_is_alive(const struct jsonrpc_session *s) +{ + return s->rpc || s->stream || reconnect_get_max_tries(s->reconnect); +} + +bool +jsonrpc_session_is_connected(const struct jsonrpc_session *s) +{ + return s->rpc != NULL; +} + +unsigned int +jsonrpc_session_get_seqno(const struct jsonrpc_session *s) +{ + return s->seqno; +} + +int +jsonrpc_session_get_status(const struct jsonrpc_session *s) +{ + return s && s->rpc ? jsonrpc_get_status(s->rpc) : 0; +} + +void +jsonrpc_session_get_reconnect_stats(const struct jsonrpc_session *s, + struct reconnect_stats *stats) +{ + reconnect_get_stats(s->reconnect, time_msec(), stats); +} + +void +jsonrpc_session_force_reconnect(struct jsonrpc_session *s) +{ + reconnect_force_reconnect(s->reconnect, time_msec()); +} + +void +jsonrpc_session_set_max_backoff(struct jsonrpc_session *s, int max_backoff) +{ + reconnect_set_backoff(s->reconnect, 0, max_backoff); +} + +void +jsonrpc_session_set_probe_interval(struct jsonrpc_session *s, + int probe_interval) +{ + reconnect_set_probe_interval(s->reconnect, probe_interval); +}