X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=afcc5208242936e58d51972833891653991a8714;hb=ccf16467108326ba35686f549534f254db02b113;hp=828bdac21a90e6eebe8b9ac62c3fb72fd42ca1f4;hpb=b302749b70bd8b487bdfe0ffcdacbf730da8b6ca;p=sliver-openvswitch.git diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c index 828bdac21..afcc52082 100644 --- a/lib/jsonrpc.c +++ b/lib/jsonrpc.c @@ -28,13 +28,12 @@ #include "list.h" #include "ofpbuf.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; @@ -47,7 +46,7 @@ struct jsonrpc { struct jsonrpc_msg *received; /* Output. */ - struct ovs_queue output; + struct list output; /* Contains "struct ofpbuf"s. */ size_t backlog; }; @@ -86,7 +85,7 @@ 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; } @@ -109,8 +108,8 @@ 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); @@ -118,7 +117,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) { @@ -136,12 +136,18 @@ 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); } } } +/* + * Possible status values: + * - 0: no error yet + * - >0: errno value + * - EOF: end of file (remote end closed connection; not necessarily an error) + */ int jsonrpc_get_status(const struct jsonrpc *rpc) { @@ -215,10 +221,10 @@ 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; @@ -249,7 +255,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; } @@ -267,6 +272,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; } } @@ -303,7 +314,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); @@ -407,7 +418,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; } @@ -695,7 +706,10 @@ jsonrpc_session_open(const char *name) } /* 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) { @@ -703,6 +717,7 @@ jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc) 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()); @@ -796,7 +811,7 @@ jsonrpc_session_run(struct jsonrpc_session *s) jsonrpc_run(s->rpc); 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); } } else if (s->stream) { @@ -861,6 +876,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) { @@ -894,8 +911,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; @@ -932,8 +949,34 @@ 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); +}