X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fstream.c;h=da089ae4b916df080b924ba8b7f412ad1591398f;hb=4b570f12766f3c4eeb527de69d8eedfd59c34b86;hp=2349b0c116be300ce1cb7953bbf5bde50c111628;hpb=f39dc942afd5fe241903aada30850a1d96122c8c;p=sliver-openvswitch.git diff --git a/lib/stream.c b/lib/stream.c index 2349b0c11..da089ae4b 100644 --- a/lib/stream.c +++ b/lib/stream.c @@ -1,5 +1,5 @@ /* - * Copyright (c) 2008, 2009, 2010 Nicira Networks. + * Copyright (c) 2008, 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. @@ -16,7 +16,6 @@ #include #include "stream-provider.h" -#include #include #include #include @@ -25,6 +24,7 @@ #include #include "coverage.h" #include "dynamic-string.h" +#include "fatal-signal.h" #include "flow.h" #include "ofp-print.h" #include "ofpbuf.h" @@ -34,10 +34,13 @@ #include "poll-loop.h" #include "random.h" #include "util.h" - -#define THIS_MODULE VLM_stream #include "vlog.h" +VLOG_DEFINE_THIS_MODULE(stream); + +COVERAGE_DEFINE(pstream_open); +COVERAGE_DEFINE(stream_open); + /* State of an active stream.*/ enum stream_state { SCS_CONNECTING, /* Underlying stream is not connected. */ @@ -45,7 +48,7 @@ enum stream_state { SCS_DISCONNECTED /* Connection failed or connection closed. */ }; -static struct stream_class *stream_classes[] = { +static const struct stream_class *stream_classes[] = { &tcp_stream_class, &unix_stream_class, #ifdef HAVE_OPENSSL @@ -53,7 +56,7 @@ static struct stream_class *stream_classes[] = { #endif }; -static struct pstream_class *pstream_classes[] = { +static const struct pstream_class *pstream_classes[] = { &ptcp_pstream_class, &punix_pstream_class, #ifdef HAVE_OPENSSL @@ -69,28 +72,28 @@ check_stream_classes(void) size_t i; for (i = 0; i < ARRAY_SIZE(stream_classes); i++) { - struct stream_class *class = stream_classes[i]; - assert(class->name != NULL); - assert(class->open != NULL); + const struct stream_class *class = stream_classes[i]; + ovs_assert(class->name != NULL); + ovs_assert(class->open != NULL); if (class->close || class->recv || class->send || class->run || class->run_wait || class->wait) { - assert(class->close != NULL); - assert(class->recv != NULL); - assert(class->send != NULL); - assert(class->wait != NULL); + ovs_assert(class->close != NULL); + ovs_assert(class->recv != NULL); + ovs_assert(class->send != NULL); + ovs_assert(class->wait != NULL); } else { /* This class delegates to another one. */ } } for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) { - struct pstream_class *class = pstream_classes[i]; - assert(class->name != NULL); - assert(class->listen != NULL); + const struct pstream_class *class = pstream_classes[i]; + ovs_assert(class->name != NULL); + ovs_assert(class->listen != NULL); if (class->close || class->accept || class->wait) { - assert(class->close != NULL); - assert(class->accept != NULL); - assert(class->wait != NULL); + ovs_assert(class->close != NULL); + ovs_assert(class->accept != NULL); + ovs_assert(class->wait != NULL); } else { /* This class delegates to another one. */ } @@ -150,7 +153,7 @@ stream_usage(const char *name, bool active, bool passive, * a null pointer into '*classp' if 'name' is in the wrong form or if no such * class exists. */ static int -stream_lookup_class(const char *name, struct stream_class **classp) +stream_lookup_class(const char *name, const struct stream_class **classp) { size_t prefix_len; size_t i; @@ -163,7 +166,7 @@ stream_lookup_class(const char *name, struct stream_class **classp) return EAFNOSUPPORT; } for (i = 0; i < ARRAY_SIZE(stream_classes); i++) { - struct stream_class *class = stream_classes[i]; + const struct stream_class *class = stream_classes[i]; if (strlen(class->name) == prefix_len && !memcmp(class->name, name, prefix_len)) { *classp = class; @@ -178,7 +181,7 @@ stream_lookup_class(const char *name, struct stream_class **classp) int stream_verify_name(const char *name) { - struct stream_class *class; + const struct stream_class *class; return stream_lookup_class(name, &class); } @@ -190,9 +193,9 @@ stream_verify_name(const char *name) * stores a pointer to the new connection in '*streamp', otherwise a null * pointer. */ int -stream_open(const char *name, struct stream **streamp) +stream_open(const char *name, struct stream **streamp, uint8_t dscp) { - struct stream_class *class; + const struct stream_class *class; struct stream *stream; char *suffix_copy; int error; @@ -207,7 +210,7 @@ stream_open(const char *name, struct stream **streamp) /* Call class's "open" function. */ suffix_copy = xstrdup(strchr(name, ':') + 1); - error = class->open(name, suffix_copy, &stream); + error = class->open(name, suffix_copy, &stream, dscp); free(suffix_copy); if (error) { goto error; @@ -237,14 +240,18 @@ stream_open_block(int error, struct stream **streamp) { struct stream *stream = *streamp; - while (error == EAGAIN) { - stream_run(stream); - stream_run_wait(stream); - stream_connect_wait(stream); - poll_block(); - error = stream_connect(stream); - assert(error != EINPROGRESS); + fatal_signal_run(); + + if (!error) { + while ((error = stream_connect(stream)) == EAGAIN) { + stream_run(stream); + stream_run_wait(stream); + stream_connect_wait(stream); + poll_block(); + } + ovs_assert(error != EINPROGRESS); } + if (error) { stream_close(stream); *streamp = NULL; @@ -275,7 +282,7 @@ stream_get_name(const struct stream *stream) /* Returns the IP address of the peer, or 0 if the peer is not connected over * an IP-based protocol or if its IP address is not yet known. */ -uint32_t +ovs_be32 stream_get_remote_ip(const struct stream *stream) { return stream->remote_ip; @@ -283,7 +290,7 @@ stream_get_remote_ip(const struct stream *stream) /* Returns the transport port of the peer, or 0 if the connection does not * contain a port or if the port is not yet known. */ -uint16_t +ovs_be16 stream_get_remote_port(const struct stream *stream) { return stream->remote_port; @@ -291,7 +298,7 @@ stream_get_remote_port(const struct stream *stream) /* Returns the IP address used to connect to the peer, or 0 if the connection * is not an IP-based protocol or if its IP address is not yet known. */ -uint32_t +ovs_be32 stream_get_local_ip(const struct stream *stream) { return stream->local_ip; @@ -299,7 +306,7 @@ stream_get_local_ip(const struct stream *stream) /* Returns the transport port used to connect to the peer, or 0 if the * connection does not contain a port or if the port is not yet known. */ -uint16_t +ovs_be16 stream_get_local_port(const struct stream *stream) { return stream->local_port; @@ -309,7 +316,7 @@ static void scs_connecting(struct stream *stream) { int retval = (stream->class->connect)(stream); - assert(retval != EINPROGRESS); + ovs_assert(retval != EINPROGRESS); if (!retval) { stream->state = SCS_CONNECTED; } else if (retval != EAGAIN) { @@ -318,10 +325,10 @@ scs_connecting(struct stream *stream) } } -/* Tries to complete the connection on 'stream', which must be an active - * stream. If 'stream''s connection is complete, returns 0 if the connection - * was successful or a positive errno value if it failed. If the - * connection is still in progress, returns EAGAIN. */ +/* Tries to complete the connection on 'stream'. If 'stream''s connection is + * complete, returns 0 if the connection was successful or a positive errno + * value if it failed. If the connection is still in progress, returns + * EAGAIN. */ int stream_connect(struct stream *stream) { @@ -411,8 +418,8 @@ stream_run_wait(struct stream *stream) void stream_wait(struct stream *stream, enum stream_wait_type wait) { - assert(wait == STREAM_CONNECT || wait == STREAM_RECV - || wait == STREAM_SEND); + ovs_assert(wait == STREAM_CONNECT || wait == STREAM_RECV + || wait == STREAM_SEND); switch (stream->state) { case SCS_CONNECTING: @@ -449,7 +456,7 @@ stream_send_wait(struct stream *stream) * a null pointer into '*classp' if 'name' is in the wrong form or if no such * class exists. */ static int -pstream_lookup_class(const char *name, struct pstream_class **classp) +pstream_lookup_class(const char *name, const struct pstream_class **classp) { size_t prefix_len; size_t i; @@ -462,7 +469,7 @@ pstream_lookup_class(const char *name, struct pstream_class **classp) return EAFNOSUPPORT; } for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) { - struct pstream_class *class = pstream_classes[i]; + const struct pstream_class *class = pstream_classes[i]; if (strlen(class->name) == prefix_len && !memcmp(class->name, name, prefix_len)) { *classp = class; @@ -477,10 +484,29 @@ pstream_lookup_class(const char *name, struct pstream_class **classp) int pstream_verify_name(const char *name) { - struct pstream_class *class; + const struct pstream_class *class; return pstream_lookup_class(name, &class); } +/* Returns 1 if the stream or pstream specified by 'name' needs periodic probes + * to verify connectivity. For [p]streams which need probes, it can take a + * long time to notice the connection has been dropped. Returns 0 if the + * stream or pstream does not need probes, and -1 if 'name' is not valid. */ +int +stream_or_pstream_needs_probes(const char *name) +{ + const struct pstream_class *pclass; + const struct stream_class *class; + + if (!stream_lookup_class(name, &class)) { + return class->needs_probes; + } else if (!pstream_lookup_class(name, &pclass)) { + return pclass->needs_probes; + } else { + return -1; + } +} + /* Attempts to start listening for remote stream connections. 'name' is a * connection name in the form "TYPE:ARGS", where TYPE is an passive stream * class's name and ARGS are stream class-specific. @@ -489,9 +515,9 @@ pstream_verify_name(const char *name) * stores a pointer to the new connection in '*pstreamp', otherwise a null * pointer. */ int -pstream_open(const char *name, struct pstream **pstreamp) +pstream_open(const char *name, struct pstream **pstreamp, uint8_t dscp) { - struct pstream_class *class; + const struct pstream_class *class; struct pstream *pstream; char *suffix_copy; int error; @@ -506,7 +532,7 @@ pstream_open(const char *name, struct pstream **pstreamp) /* Call class's "open" function. */ suffix_copy = xstrdup(strchr(name, ':') + 1); - error = class->listen(name, suffix_copy, &pstream); + error = class->listen(name, suffix_copy, &pstream, dscp); free(suffix_copy); if (error) { goto error; @@ -553,8 +579,8 @@ pstream_accept(struct pstream *pstream, struct stream **new_stream) if (retval) { *new_stream = NULL; } else { - assert((*new_stream)->state != SCS_CONNECTING - || (*new_stream)->class->connect); + ovs_assert((*new_stream)->state != SCS_CONNECTING + || (*new_stream)->class->connect); } return retval; } @@ -570,6 +596,7 @@ pstream_accept_block(struct pstream *pstream, struct stream **new_stream) { int error; + fatal_signal_run(); while ((error = pstream_accept(pstream, new_stream)) == EAGAIN) { pstream_wait(pstream); poll_block(); @@ -585,6 +612,23 @@ pstream_wait(struct pstream *pstream) { (pstream->class->wait)(pstream); } + +int +pstream_set_dscp(struct pstream *pstream, uint8_t dscp) +{ + if (pstream->class->set_dscp) { + return pstream->class->set_dscp(pstream, dscp); + } + return 0; +} + +/* Returns the transport port on which 'pstream' is listening, or 0 if the + * concept doesn't apply. */ +ovs_be16 +pstream_get_bound_port(const struct pstream *pstream) +{ + return pstream->bound_port; +} /* Initializes 'stream' as a new stream named 'name', implemented via 'class'. * The initial connection status, supplied as 'connect_status', is interpreted @@ -604,49 +648,57 @@ pstream_wait(struct pstream *pstream) * * The caller retains ownership of 'name'. */ void -stream_init(struct stream *stream, struct stream_class *class, +stream_init(struct stream *stream, const struct stream_class *class, int connect_status, const char *name) { + memset(stream, 0, sizeof *stream); stream->class = class; stream->state = (connect_status == EAGAIN ? SCS_CONNECTING : !connect_status ? SCS_CONNECTED : SCS_DISCONNECTED); stream->error = connect_status; stream->name = xstrdup(name); - assert(stream->state != SCS_CONNECTING || class->connect); + ovs_assert(stream->state != SCS_CONNECTING || class->connect); } void -stream_set_remote_ip(struct stream *stream, uint32_t ip) +stream_set_remote_ip(struct stream *stream, ovs_be32 ip) { stream->remote_ip = ip; } void -stream_set_remote_port(struct stream *stream, uint16_t port) +stream_set_remote_port(struct stream *stream, ovs_be16 port) { stream->remote_port = port; } void -stream_set_local_ip(struct stream *stream, uint32_t ip) +stream_set_local_ip(struct stream *stream, ovs_be32 ip) { stream->local_ip = ip; } void -stream_set_local_port(struct stream *stream, uint16_t port) +stream_set_local_port(struct stream *stream, ovs_be16 port) { stream->local_port = port; } void -pstream_init(struct pstream *pstream, struct pstream_class *class, +pstream_init(struct pstream *pstream, const struct pstream_class *class, const char *name) { + memset(pstream, 0, sizeof *pstream); pstream->class = class; pstream->name = xstrdup(name); } + +void +pstream_set_bound_port(struct pstream *pstream, ovs_be16 port) +{ + pstream->bound_port = port; +} static int count_fields(const char *s_) @@ -672,7 +724,8 @@ int stream_open_with_default_ports(const char *name_, uint16_t default_tcp_port, uint16_t default_ssl_port, - struct stream **streamp) + struct stream **streamp, + uint8_t dscp) { char *name; int error; @@ -684,7 +737,7 @@ stream_open_with_default_ports(const char *name_, } else { name = xstrdup(name_); } - error = stream_open(name, streamp); + error = stream_open(name, streamp, dscp); free(name); return error; @@ -697,7 +750,8 @@ int pstream_open_with_default_ports(const char *name_, uint16_t default_ptcp_port, uint16_t default_pssl_port, - struct pstream **pstreamp) + struct pstream **pstreamp, + uint8_t dscp) { char *name; int error; @@ -709,10 +763,91 @@ pstream_open_with_default_ports(const char *name_, } else { name = xstrdup(name_); } - error = pstream_open(name, pstreamp); + error = pstream_open(name, pstreamp, dscp); free(name); return error; } +/* + * This function extracts IP address and port from the target string. + * + * - On success, function returns true and fills *sin structure with port + * and IP address. If port was absent in target string then it will use + * corresponding default port value. + * - On error, function returns false and *sin contains garbage. + */ +bool +stream_parse_target_with_default_ports(const char *target, + uint16_t default_tcp_port, + uint16_t default_ssl_port, + struct sockaddr_in *sin) +{ + return (!strncmp(target, "tcp:", 4) + && inet_parse_active(target + 4, default_tcp_port, sin)) || + (!strncmp(target, "ssl:", 4) + && inet_parse_active(target + 4, default_ssl_port, sin)); +} + +/* Attempts to guess the content type of a stream whose first few bytes were + * the 'size' bytes of 'data'. */ +static enum stream_content_type +stream_guess_content(const uint8_t *data, ssize_t size) +{ + if (size >= 2) { +#define PAIR(A, B) (((A) << 8) | (B)) + switch (PAIR(data[0], data[1])) { + case PAIR(0x16, 0x03): /* Handshake, version 3. */ + return STREAM_SSL; + case PAIR('{', '"'): + return STREAM_JSONRPC; + case PAIR(OFP10_VERSION, 0 /* OFPT_HELLO */): + return STREAM_OPENFLOW; + } + } + + return STREAM_UNKNOWN; +} + +/* Returns a string represenation of 'type'. */ +static const char * +stream_content_type_to_string(enum stream_content_type type) +{ + switch (type) { + case STREAM_UNKNOWN: + default: + return "unknown"; + + case STREAM_JSONRPC: + return "JSON-RPC"; + + case STREAM_OPENFLOW: + return "OpenFlow"; + case STREAM_SSL: + return "SSL"; + } +} + +/* Attempts to guess the content type of a stream whose first few bytes were + * the 'size' bytes of 'data'. If this is done successfully, and the guessed + * content type is other than 'expected_type', then log a message in vlog + * module 'module', naming 'stream_name' as the source, explaining what + * content was expected and what was actually received. */ +void +stream_report_content(const void *data, ssize_t size, + enum stream_content_type expected_type, + struct vlog_module *module, const char *stream_name) +{ + static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5); + enum stream_content_type actual_type; + + actual_type = stream_guess_content(data, size); + if (actual_type != expected_type && actual_type != STREAM_UNKNOWN) { + vlog_rate_limit(module, VLL_WARN, &rl, + "%s: received %s data on %s channel", + stream_name, + stream_content_type_to_string(actual_type), + stream_content_type_to_string(expected_type)); + } +}