X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=1ea5398b44068a53070f20de0a1064026f1fe4a7;hb=fba6bd1d3f5891471daea8bf5da22303c2d889df;hp=764677114d4fa1ed32046d3bf4faaba28e56ea0a;hpb=78fdd76d907f62255b3db02ea90e540f4a2b006b;p=sliver-openvswitch.git diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c index 764677114..1ea5398b4 100644 --- a/lib/jsonrpc.c +++ b/lib/jsonrpc.c @@ -1,5 +1,5 @@ /* - * Copyright (c) 2009, 2010, 2011, 2012 Nicira Networks. + * Copyright (c) 2009, 2010, 2011, 2012, 2013 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,7 +18,6 @@ #include "jsonrpc.h" -#include #include #include "byteq.h" @@ -60,19 +59,20 @@ 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) +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); + JSONRPC_SSL_PORT, streamp, + dscp); } /* 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) +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); + JSONRPC_SSL_PORT, pstreamp, dscp); } /* Returns a new JSON-RPC stream that uses 'stream' for input and output. The @@ -82,7 +82,7 @@ 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)); @@ -177,6 +177,14 @@ 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 * @@ -276,13 +284,19 @@ jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg) int jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) { + int i; + *msgp = NULL; if (rpc->status) { return rpc->status; } - while (!rpc->received) { - if (byteq_is_empty(&rpc->input)) { + 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 chunk; int retval; @@ -327,9 +341,7 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp) } } - *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 @@ -412,9 +424,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); @@ -458,7 +472,7 @@ jsonrpc_received(struct jsonrpc *rpc) static void jsonrpc_error(struct jsonrpc *rpc, int error) { - assert(error); + ovs_assert(error); if (!rpc->status) { rpc->status = error; jsonrpc_cleanup(rpc); @@ -730,21 +744,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; @@ -756,9 +776,18 @@ 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; @@ -770,7 +799,7 @@ jsonrpc_session_open(const char *name) * 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; @@ -780,6 +809,7 @@ jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc) 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; @@ -823,12 +853,15 @@ jsonrpc_session_connect(struct jsonrpc_session *s) jsonrpc_session_disconnect(s); if (!reconnect_is_passive(s->reconnect)) { - error = jsonrpc_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 : jsonrpc_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()); } @@ -865,13 +898,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(), error); jsonrpc_session_disconnect(s); + s->last_error = error; } } else if (s->stream) { int error; @@ -959,10 +1007,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; @@ -1014,6 +1073,12 @@ 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) @@ -1039,3 +1104,27 @@ jsonrpc_session_set_probe_interval(struct jsonrpc_session *s, { 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), 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); + } +}