X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=842d1173a9dfcec335ad989258a18dadc8b299b3;hb=HEAD;hp=1ea5398b44068a53070f20de0a1064026f1fe4a7;hpb=780325b5b8d4c0552b4b7719e0a38200d99f6b08;p=sliver-openvswitch.git diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c index 1ea5398b4..842d1173a 100644 --- a/lib/jsonrpc.c +++ b/lib/jsonrpc.c @@ -1,5 +1,5 @@ /* - * Copyright (c) 2009, 2010, 2011, 2012, 2013 Nicira, Inc. + * 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. @@ -26,6 +26,7 @@ #include "json.h" #include "list.h" #include "ofpbuf.h" +#include "ovs-thread.h" #include "poll-loop.h" #include "reconnect.h" #include "stream.h" @@ -41,8 +42,8 @@ struct jsonrpc { /* Input. */ struct byteq input; + uint8_t input_buffer[512]; struct json_parser *parser; - struct jsonrpc_msg *received; /* Output. */ struct list output; /* Contains "struct ofpbuf"s. */ @@ -52,27 +53,26 @@ struct jsonrpc { /* 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 ports if none is specified. */ + * JSONRPC port if none is specified. */ int jsonrpc_stream_open(const char *name, struct stream **streamp, uint8_t dscp) { - return stream_open_with_default_ports(name, JSONRPC_TCP_PORT, - JSONRPC_SSL_PORT, streamp, - 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 ports if none is specified. */ + * JSONRPC port if none is specified. */ int jsonrpc_pstream_open(const char *name, struct pstream **pstreamp, uint8_t dscp) { - return pstream_open_with_default_ports(name, JSONRPC_TCP_PORT, - JSONRPC_SSL_PORT, pstreamp, 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 @@ -87,7 +87,7 @@ jsonrpc_open(struct stream *stream) rpc = xzalloc(sizeof *rpc); rpc->name = xstrdup(stream_get_name(stream)); rpc->stream = stream; - byteq_init(&rpc->input); + byteq_init(&rpc->input, rpc->input_buffer, sizeof rpc->input_buffer); list_init(&rpc->output); return rpc; @@ -118,18 +118,18 @@ jsonrpc_run(struct jsonrpc *rpc) 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) { + 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; @@ -255,7 +255,7 @@ jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg) buf = xmalloc(sizeof *buf); ofpbuf_use(buf, s, length); - buf->size = length; + ofpbuf_set_size(buf, length); list_push_back(&rpc->output, &buf->list_node); rpc->backlog += length; @@ -292,11 +292,10 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) } for (i = 0; i < 50; i++) { - if (rpc->received) { - *msgp = rpc->received; - rpc->received = NULL; - return 0; - } else if (byteq_is_empty(&rpc->input)) { + size_t n, used; + + /* Fill our input buffer if it's empty. */ + if (byteq_is_empty(&rpc->input)) { size_t chunk; int retval; @@ -307,7 +306,7 @@ 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; } @@ -316,27 +315,31 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) 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) { - 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; + + 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; } } } @@ -349,8 +352,8 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) void jsonrpc_recv_wait(struct jsonrpc *rpc) { - if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) { - (poll_immediate_wake)(rpc->name); + if (rpc->status || !byteq_is_empty(&rpc->input)) { + poll_immediate_wake_at(rpc->name); } else { stream_recv_wait(rpc->stream); } @@ -439,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; @@ -453,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); @@ -462,11 +468,11 @@ 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; } static void @@ -488,9 +494,6 @@ jsonrpc_cleanup(struct jsonrpc *rpc) json_parser_abort(rpc->parser); rpc->parser = NULL; - jsonrpc_msg_destroy(rpc->received); - rpc->received = NULL; - ofpbuf_list_delete(&rpc->output); rpc->backlog = 0; } @@ -513,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 * @@ -1086,6 +1092,14 @@ jsonrpc_session_get_reconnect_stats(const struct jsonrpc_session *s, 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) { @@ -1116,7 +1130,8 @@ jsonrpc_session_set_dscp(struct jsonrpc_session *s, error = pstream_set_dscp(s->pstream, dscp); if (error) { VLOG_ERR("%s: failed set_dscp %s", - reconnect_get_name(s->reconnect), strerror(error)); + reconnect_get_name(s->reconnect), + ovs_strerror(error)); } /* * XXX race window between setting dscp to listening socket