X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=842d1173a9dfcec335ad989258a18dadc8b299b3;hb=HEAD;hp=f0ac27bc94045d4f04623587e764a50a7521b034;hpb=c9f3f37a9d3eeac3740122569ec96c3658b615df;p=sliver-openvswitch.git diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c index f0ac27bc9..842d1173a 100644 --- a/lib/jsonrpc.c +++ b/lib/jsonrpc.c @@ -1,5 +1,5 @@ /* - * Copyright (c) 2009, 2010 Nicira Networks. + * Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014 Nicira, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,22 +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 "ovs-thread.h" #include "poll-loop.h" -#include "queue.h" #include "reconnect.h" #include "stream.h" #include "timeval.h" - -#define THIS_MODULE VLM_jsonrpc #include "vlog.h" + +VLOG_DEFINE_THIS_MODULE(jsonrpc); struct jsonrpc { struct stream *stream; @@ -42,36 +42,59 @@ struct jsonrpc { /* Input. */ struct byteq input; + uint8_t input_buffer[512]; struct json_parser *parser; - struct jsonrpc_msg *received; /* Output. */ - struct ovs_queue output; + struct list output; /* Contains "struct ofpbuf"s. */ size_t backlog; }; /* Rate limit for error messages. */ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5); -static void jsonrpc_received(struct jsonrpc *); +static struct jsonrpc_msg *jsonrpc_parse_received_message(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 port if none is specified. */ +int +jsonrpc_stream_open(const char *name, struct stream **streamp, uint8_t dscp) +{ + return stream_open_with_default_port(name, OVSDB_OLD_PORT, + streamp, dscp); +} +/* This is just the same as pstream_open() except that it uses the default + * JSONRPC port if none is specified. */ +int +jsonrpc_pstream_open(const char *name, struct pstream **pstreamp, uint8_t dscp) +{ + return pstream_open_with_default_port(name, OVSDB_OLD_PORT, + pstreamp, dscp); +} + +/* 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) { struct jsonrpc *rpc; - assert(stream != NULL); + ovs_assert(stream != NULL); rpc = xzalloc(sizeof *rpc); rpc->name = xstrdup(stream_get_name(stream)); rpc->stream = stream; - byteq_init(&rpc->input); - queue_init(&rpc->output); + byteq_init(&rpc->input, rpc->input_buffer, sizeof rpc->input_buffer); + 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) { @@ -82,6 +105,7 @@ jsonrpc_close(struct jsonrpc *rpc) } } +/* Performs periodic maintenance on 'rpc', such as flushing output buffers. */ void jsonrpc_run(struct jsonrpc *rpc) { @@ -90,21 +114,22 @@ jsonrpc_run(struct jsonrpc *rpc) } stream_run(rpc->stream); - while (!queue_is_empty(&rpc->output)) { - struct ofpbuf *buf = rpc->output.head; + 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); + retval = stream_send(rpc->stream, ofpbuf_data(buf), ofpbuf_size(buf)); if (retval >= 0) { rpc->backlog -= retval; ofpbuf_pull(buf, retval); - if (!buf->size) { - ofpbuf_delete(queue_pop_head(&rpc->output)); + if (!ofpbuf_size(buf)) { + list_remove(&buf->list_node); + ofpbuf_delete(buf); } } else { if (retval != -EAGAIN) { VLOG_WARN_RL(&rl, "%s: send error: %s", - rpc->name, strerror(-retval)); + rpc->name, ovs_strerror(-retval)); jsonrpc_error(rpc, -retval); } break; @@ -112,29 +137,56 @@ 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) { stream_run_wait(rpc->stream); - if (!queue_is_empty(&rpc->output)) { + 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 the number of bytes that have been received on 'rpc''s underlying + * stream. (The value wraps around if it exceeds UINT_MAX.) */ +unsigned int +jsonrpc_get_received_bytes(const struct jsonrpc *rpc) +{ + return rpc->input.head; +} + +/* 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) { @@ -172,7 +224,15 @@ jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title, } } -/* Always takes ownership of 'msg', regardless of success. */ +/* 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) { @@ -195,25 +255,46 @@ 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); + ofpbuf_set_size(buf, length); + 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) { + int i; + *msgp = NULL; if (rpc->status) { return rpc->status; } - while (!rpc->received) { + for (i = 0; i < 50; i++) { + size_t n, used; + + /* Fill our input buffer if it's empty. */ if (byteq_is_empty(&rpc->input)) { size_t chunk; int retval; @@ -225,56 +306,71 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) return EAGAIN; } else { VLOG_WARN_RL(&rl, "%s: receive error: %s", - rpc->name, strerror(-retval)); + rpc->name, ovs_strerror(-retval)); jsonrpc_error(rpc, -retval); return rpc->status; } } else if (retval == 0) { - VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name); jsonrpc_error(rpc, EOF); return EOF; } byteq_advance_head(&rpc->input, retval); - } else { - size_t n, used; + } - if (!rpc->parser) { - rpc->parser = json_parser_create(0); + /* We have some input. Feed it into the JSON parser. */ + if (!rpc->parser) { + rpc->parser = json_parser_create(0); + } + n = byteq_tailroom(&rpc->input); + used = json_parser_feed(rpc->parser, + (char *) byteq_tail(&rpc->input), n); + byteq_advance_tail(&rpc->input, used); + + /* If we have complete JSON, attempt to parse it as JSON-RPC. */ + if (json_parser_is_done(rpc->parser)) { + *msgp = jsonrpc_parse_received_message(rpc); + if (*msgp) { + return 0; } - n = byteq_tailroom(&rpc->input); - used = json_parser_feed(rpc->parser, - (char *) byteq_tail(&rpc->input), n); - byteq_advance_tail(&rpc->input, used); - if (json_parser_is_done(rpc->parser)) { - jsonrpc_received(rpc); - if (rpc->status) { - return rpc->status; + + if (rpc->status) { + const struct byteq *q = &rpc->input; + if (q->head <= q->size) { + stream_report_content(q->buffer, q->head, STREAM_JSONRPC, + THIS_MODULE, rpc->name); } + return rpc->status; } } } - *msgp = rpc->received; - rpc->received = NULL; - return 0; + return EAGAIN; } +/* 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(); + if (rpc->status || !byteq_is_empty(&rpc->input)) { + poll_immediate_wake_at(rpc->name); } else { stream_recv_wait(rpc->stream); } } -/* Always takes ownership of 'msg', regardless of success. */ +/* 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; @@ -282,7 +378,7 @@ jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg) for (;;) { jsonrpc_run(rpc); - if (queue_is_empty(&rpc->output) || rpc->status) { + if (list_is_empty(&rpc->output) || rpc->status) { return rpc->status; } jsonrpc_wait(rpc); @@ -290,12 +386,15 @@ jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg) } } +/* 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; } @@ -306,7 +405,15 @@ jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) } } -/* Always takes ownership of 'request', regardless of success. */ +/* 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) @@ -320,9 +427,11 @@ jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request, if (!error) { for (;;) { error = jsonrpc_recv_block(rpc, &reply); - if (error - || (reply->type == JSONRPC_REPLY - && json_equal(id, reply->id))) { + if (error) { + break; + } + if ((reply->type == JSONRPC_REPLY || reply->type == JSONRPC_ERROR) + && json_equal(id, reply->id)) { break; } jsonrpc_msg_destroy(reply); @@ -333,8 +442,11 @@ jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request, return error; } -static void -jsonrpc_received(struct jsonrpc *rpc) +/* Attempts to parse the content of 'rpc->parser' (which is complete JSON) as a + * JSON-RPC message. If successful, returns the JSON-RPC message. On failure, + * signals an error on 'rpc' with jsonrpc_error() and returns NULL. */ +static struct jsonrpc_msg * +jsonrpc_parse_received_message(struct jsonrpc *rpc) { struct jsonrpc_msg *msg; struct json *json; @@ -347,7 +459,7 @@ jsonrpc_received(struct jsonrpc *rpc) rpc->name, json_string(json)); jsonrpc_error(rpc, EPROTO); json_destroy(json); - return; + return NULL; } error = jsonrpc_msg_from_json(json, &msg); @@ -356,17 +468,17 @@ jsonrpc_received(struct jsonrpc *rpc) rpc->name, error); free(error); jsonrpc_error(rpc, EPROTO); - return; + return NULL; } jsonrpc_log_msg(rpc, "received", msg); - rpc->received = msg; + return msg; } -void +static void jsonrpc_error(struct jsonrpc *rpc, int error) { - assert(error); + ovs_assert(error); if (!rpc->status) { rpc->status = error; jsonrpc_cleanup(rpc); @@ -382,10 +494,7 @@ jsonrpc_cleanup(struct jsonrpc *rpc) json_parser_abort(rpc->parser); rpc->parser = NULL; - jsonrpc_msg_destroy(rpc->received); - rpc->received = NULL; - - queue_clear(&rpc->output); + ofpbuf_list_delete(&rpc->output); rpc->backlog = 0; } @@ -407,8 +516,11 @@ jsonrpc_create(enum jsonrpc_msg_type type, const char *method, static struct json * jsonrpc_create_id(void) { - static unsigned int id; - return json_integer_create(id++); + static atomic_uint next_id = ATOMIC_VAR_INIT(0); + unsigned int id; + + atomic_add(&next_id, 1, &id); + return json_integer_create(id); } struct jsonrpc_msg * @@ -638,21 +750,27 @@ struct jsonrpc_session { struct jsonrpc *rpc; struct stream *stream; struct pstream *pstream; + int last_error; unsigned int seqno; + uint8_t dscp; }; /* 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'. + * jsonrpc_session connects to 'name'. If 'retry' is true, then the new + * session connects and reconnects to 'name', with backoff. If 'retry' is + * false, the new session will only try to connect once and after a connection + * failure or a disconnection jsonrpc_session_is_alive() will return false for + * the new session. * * 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) +jsonrpc_session_open(const char *name, bool retry) { struct jsonrpc_session *s; @@ -664,26 +782,40 @@ jsonrpc_session_open(const char *name) s->stream = NULL; s->pstream = NULL; s->seqno = 0; + s->dscp = 0; + s->last_error = 0; if (!pstream_verify_name(name)) { reconnect_set_passive(s->reconnect, true, time_msec()); + } else if (!retry) { + reconnect_set_max_tries(s->reconnect, 1); + reconnect_set_backoff(s->reconnect, INT_MAX, INT_MAX); + } + + if (!stream_or_pstream_needs_probes(name)) { + reconnect_set_probe_interval(s->reconnect, 0); } return s; } /* Creates and returns a jsonrpc_session that is initially connected to - * 'jsonrpc'. If the connection is dropped, it will not be reconnected. */ + * '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) +jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc, uint8_t dscp) { 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->dscp = dscp; s->rpc = jsonrpc; s->stream = NULL; s->pstream = NULL; @@ -727,12 +859,15 @@ jsonrpc_session_connect(struct jsonrpc_session *s) jsonrpc_session_disconnect(s); if (!reconnect_is_passive(s->reconnect)) { - error = stream_open(name, &s->stream); + error = jsonrpc_stream_open(name, &s->stream, s->dscp); if (!error) { reconnect_connecting(s->reconnect, time_msec()); + } else { + s->last_error = error; } } else { - error = s->pstream ? 0 : pstream_open(name, &s->pstream); + error = s->pstream ? 0 : jsonrpc_pstream_open(name, &s->pstream, + s->dscp); if (!error) { reconnect_listening(s->reconnect, time_msec()); } @@ -769,13 +904,28 @@ jsonrpc_session_run(struct jsonrpc_session *s) } if (s->rpc) { + size_t backlog; int error; + backlog = jsonrpc_get_backlog(s->rpc); jsonrpc_run(s->rpc); + if (jsonrpc_get_backlog(s->rpc) < backlog) { + /* Data previously caught in a queue was successfully sent (or + * there's an error, which we'll catch below.) + * + * We don't count data that is successfully sent immediately as + * activity, because there's a lot of queuing downstream from us, + * which means that we can push a lot of data into a connection + * that has stalled and won't ever recover. + */ + reconnect_activity(s->reconnect, time_msec()); + } + error = jsonrpc_get_status(s->rpc); if (error) { - reconnect_disconnected(s->reconnect, time_msec(), 0); + reconnect_disconnected(s->reconnect, time_msec(), error); jsonrpc_session_disconnect(s); + s->last_error = error; } } else if (s->stream) { int error; @@ -839,6 +989,8 @@ 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) { @@ -861,10 +1013,21 @@ struct jsonrpc_msg * jsonrpc_session_recv(struct jsonrpc_session *s) { if (s->rpc) { + unsigned int received_bytes; struct jsonrpc_msg *msg; + + received_bytes = jsonrpc_get_received_bytes(s->rpc); jsonrpc_recv(s->rpc, &msg); + if (received_bytes != jsonrpc_get_received_bytes(s->rpc)) { + /* Data was successfully received. + * + * Previously we only counted receiving a full message as activity, + * but with large messages or a slow connection that policy could + * time out the session mid-message. */ + reconnect_activity(s->reconnect, time_msec()); + } + 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; @@ -872,8 +1035,8 @@ jsonrpc_session_recv(struct jsonrpc_session *s) 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")) { + && 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; @@ -910,8 +1073,73 @@ 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; +} + +int +jsonrpc_session_get_last_error(const struct jsonrpc_session *s) +{ + return s->last_error; +} + +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_enable_reconnect(struct jsonrpc_session *s) +{ + reconnect_set_max_tries(s->reconnect, UINT_MAX); + reconnect_set_backoff(s->reconnect, RECONNECT_DEFAULT_MIN_BACKOFF, + RECONNECT_DEFAULT_MAX_BACKOFF); +} + 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); +} + +void +jsonrpc_session_set_dscp(struct jsonrpc_session *s, + uint8_t dscp) +{ + if (s->dscp != dscp) { + if (s->pstream) { + int error; + + error = pstream_set_dscp(s->pstream, dscp); + if (error) { + VLOG_ERR("%s: failed set_dscp %s", + reconnect_get_name(s->reconnect), + ovs_strerror(error)); + } + /* + * XXX race window between setting dscp to listening socket + * and accepting socket. accepted socket may have old dscp value. + * Ignore this race window for now. + */ + } + s->dscp = dscp; + jsonrpc_session_force_reconnect(s); + } +}