netdev-linux: Make queue 0 the default QOS policy
[openvswitch] / lib / stream.c
1 /*
2  * Copyright (c) 2008, 2009, 2010 Nicira Networks.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at:
7  *
8  *     http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16
17 #include <config.h>
18 #include "stream-provider.h"
19 #include <assert.h>
20 #include <errno.h>
21 #include <inttypes.h>
22 #include <netinet/in.h>
23 #include <poll.h>
24 #include <stdlib.h>
25 #include <string.h>
26 #include "coverage.h"
27 #include "dynamic-string.h"
28 #include "fatal-signal.h"
29 #include "flow.h"
30 #include "ofp-print.h"
31 #include "ofpbuf.h"
32 #include "openflow/nicira-ext.h"
33 #include "openflow/openflow.h"
34 #include "packets.h"
35 #include "poll-loop.h"
36 #include "random.h"
37 #include "util.h"
38 #include "vlog.h"
39
40 VLOG_DEFINE_THIS_MODULE(stream)
41
42 /* State of an active stream.*/
43 enum stream_state {
44     SCS_CONNECTING,             /* Underlying stream is not connected. */
45     SCS_CONNECTED,              /* Connection established. */
46     SCS_DISCONNECTED            /* Connection failed or connection closed. */
47 };
48
49 static struct stream_class *stream_classes[] = {
50     &tcp_stream_class,
51     &unix_stream_class,
52 #ifdef HAVE_OPENSSL
53     &ssl_stream_class,
54 #endif
55 };
56
57 static struct pstream_class *pstream_classes[] = {
58     &ptcp_pstream_class,
59     &punix_pstream_class,
60 #ifdef HAVE_OPENSSL
61     &pssl_pstream_class,
62 #endif
63 };
64
65 /* Check the validity of the stream class structures. */
66 static void
67 check_stream_classes(void)
68 {
69 #ifndef NDEBUG
70     size_t i;
71
72     for (i = 0; i < ARRAY_SIZE(stream_classes); i++) {
73         struct stream_class *class = stream_classes[i];
74         assert(class->name != NULL);
75         assert(class->open != NULL);
76         if (class->close || class->recv || class->send || class->run
77             || class->run_wait || class->wait) {
78             assert(class->close != NULL);
79             assert(class->recv != NULL);
80             assert(class->send != NULL);
81             assert(class->wait != NULL);
82         } else {
83             /* This class delegates to another one. */
84         }
85     }
86
87     for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) {
88         struct pstream_class *class = pstream_classes[i];
89         assert(class->name != NULL);
90         assert(class->listen != NULL);
91         if (class->close || class->accept || class->wait) {
92             assert(class->close != NULL);
93             assert(class->accept != NULL);
94             assert(class->wait != NULL);
95         } else {
96             /* This class delegates to another one. */
97         }
98     }
99 #endif
100 }
101
102 /* Prints information on active (if 'active') and passive (if 'passive')
103  * connection methods supported by the stream. */
104 void
105 stream_usage(const char *name, bool active, bool passive,
106              bool bootstrap OVS_UNUSED)
107 {
108     /* Really this should be implemented via callbacks into the stream
109      * providers, but that seems too heavy-weight to bother with at the
110      * moment. */
111
112     printf("\n");
113     if (active) {
114         printf("Active %s connection methods:\n", name);
115         printf("  tcp:IP:PORT             "
116                "PORT at remote IP\n");
117 #ifdef HAVE_OPENSSL
118         printf("  ssl:IP:PORT             "
119                "SSL PORT at remote IP\n");
120 #endif
121         printf("  unix:FILE               "
122                "Unix domain socket named FILE\n");
123     }
124
125     if (passive) {
126         printf("Passive %s connection methods:\n", name);
127         printf("  ptcp:PORT[:IP]          "
128                "listen to TCP PORT on IP\n");
129 #ifdef HAVE_OPENSSL
130         printf("  pssl:PORT[:IP]          "
131                "listen for SSL on PORT on IP\n");
132 #endif
133         printf("  punix:FILE              "
134                "listen on Unix domain socket FILE\n");
135     }
136
137 #ifdef HAVE_OPENSSL
138     printf("PKI configuration (required to use SSL):\n"
139            "  -p, --private-key=FILE  file with private key\n"
140            "  -c, --certificate=FILE  file with certificate for private key\n"
141            "  -C, --ca-cert=FILE      file with peer CA certificate\n");
142     if (bootstrap) {
143         printf("  --bootstrap-ca-cert=FILE  file with peer CA certificate "
144                "to read or create\n");
145     }
146 #endif
147 }
148
149 /* Given 'name', a stream name in the form "TYPE:ARGS", stores the class
150  * named "TYPE" into '*classp' and returns 0.  Returns EAFNOSUPPORT and stores
151  * a null pointer into '*classp' if 'name' is in the wrong form or if no such
152  * class exists. */
153 static int
154 stream_lookup_class(const char *name, struct stream_class **classp)
155 {
156     size_t prefix_len;
157     size_t i;
158
159     check_stream_classes();
160
161     *classp = NULL;
162     prefix_len = strcspn(name, ":");
163     if (name[prefix_len] == '\0') {
164         return EAFNOSUPPORT;
165     }
166     for (i = 0; i < ARRAY_SIZE(stream_classes); i++) {
167         struct stream_class *class = stream_classes[i];
168         if (strlen(class->name) == prefix_len
169             && !memcmp(class->name, name, prefix_len)) {
170             *classp = class;
171             return 0;
172         }
173     }
174     return EAFNOSUPPORT;
175 }
176
177 /* Returns 0 if 'name' is a stream name in the form "TYPE:ARGS" and TYPE is
178  * a supported stream type, otherwise EAFNOSUPPORT.  */
179 int
180 stream_verify_name(const char *name)
181 {
182     struct stream_class *class;
183     return stream_lookup_class(name, &class);
184 }
185
186 /* Attempts to connect a stream to a remote peer.  'name' is a connection name
187  * in the form "TYPE:ARGS", where TYPE is an active stream class's name and
188  * ARGS are stream class-specific.
189  *
190  * Returns 0 if successful, otherwise a positive errno value.  If successful,
191  * stores a pointer to the new connection in '*streamp', otherwise a null
192  * pointer.  */
193 int
194 stream_open(const char *name, struct stream **streamp)
195 {
196     struct stream_class *class;
197     struct stream *stream;
198     char *suffix_copy;
199     int error;
200
201     COVERAGE_INC(stream_open);
202
203     /* Look up the class. */
204     error = stream_lookup_class(name, &class);
205     if (!class) {
206         goto error;
207     }
208
209     /* Call class's "open" function. */
210     suffix_copy = xstrdup(strchr(name, ':') + 1);
211     error = class->open(name, suffix_copy, &stream);
212     free(suffix_copy);
213     if (error) {
214         goto error;
215     }
216
217     /* Success. */
218     *streamp = stream;
219     return 0;
220
221 error:
222     *streamp = NULL;
223     return error;
224 }
225
226 /* Blocks until a previously started stream connection attempt succeeds or
227  * fails.  'error' should be the value returned by stream_open() and 'streamp'
228  * should point to the stream pointer set by stream_open().  Returns 0 if
229  * successful, otherwise a positive errno value other than EAGAIN or
230  * EINPROGRESS.  If successful, leaves '*streamp' untouched; on error, closes
231  * '*streamp' and sets '*streamp' to null.
232  *
233  * Typical usage:
234  *   error = stream_open_block(stream_open("tcp:1.2.3.4:5", &stream), &stream);
235  */
236 int
237 stream_open_block(int error, struct stream **streamp)
238 {
239     struct stream *stream = *streamp;
240
241     fatal_signal_run();
242
243     if (!error) {
244         while ((error = stream_connect(stream)) == EAGAIN) {
245             stream_run(stream);
246             stream_run_wait(stream);
247             stream_connect_wait(stream);
248             poll_block();
249         }
250         assert(error != EINPROGRESS);
251     }
252
253     if (error) {
254         stream_close(stream);
255         *streamp = NULL;
256     } else {
257         *streamp = stream;
258     }
259     return error;
260 }
261
262 /* Closes 'stream'. */
263 void
264 stream_close(struct stream *stream)
265 {
266     if (stream != NULL) {
267         char *name = stream->name;
268         (stream->class->close)(stream);
269         free(name);
270     }
271 }
272
273 /* Returns the name of 'stream', that is, the string passed to
274  * stream_open(). */
275 const char *
276 stream_get_name(const struct stream *stream)
277 {
278     return stream ? stream->name : "(null)";
279 }
280
281 /* Returns the IP address of the peer, or 0 if the peer is not connected over
282  * an IP-based protocol or if its IP address is not yet known. */
283 uint32_t
284 stream_get_remote_ip(const struct stream *stream)
285 {
286     return stream->remote_ip;
287 }
288
289 /* Returns the transport port of the peer, or 0 if the connection does not
290  * contain a port or if the port is not yet known. */
291 uint16_t
292 stream_get_remote_port(const struct stream *stream)
293 {
294     return stream->remote_port;
295 }
296
297 /* Returns the IP address used to connect to the peer, or 0 if the connection
298  * is not an IP-based protocol or if its IP address is not yet known. */
299 uint32_t
300 stream_get_local_ip(const struct stream *stream)
301 {
302     return stream->local_ip;
303 }
304
305 /* Returns the transport port used to connect to the peer, or 0 if the
306  * connection does not contain a port or if the port is not yet known. */
307 uint16_t
308 stream_get_local_port(const struct stream *stream)
309 {
310     return stream->local_port;
311 }
312
313 static void
314 scs_connecting(struct stream *stream)
315 {
316     int retval = (stream->class->connect)(stream);
317     assert(retval != EINPROGRESS);
318     if (!retval) {
319         stream->state = SCS_CONNECTED;
320     } else if (retval != EAGAIN) {
321         stream->state = SCS_DISCONNECTED;
322         stream->error = retval;
323     }
324 }
325
326 /* Tries to complete the connection on 'stream'.  If 'stream''s connection is
327  * complete, returns 0 if the connection was successful or a positive errno
328  * value if it failed.  If the connection is still in progress, returns
329  * EAGAIN. */
330 int
331 stream_connect(struct stream *stream)
332 {
333     enum stream_state last_state;
334
335     do {
336         last_state = stream->state;
337         switch (stream->state) {
338         case SCS_CONNECTING:
339             scs_connecting(stream);
340             break;
341
342         case SCS_CONNECTED:
343             return 0;
344
345         case SCS_DISCONNECTED:
346             return stream->error;
347
348         default:
349             NOT_REACHED();
350         }
351     } while (stream->state != last_state);
352
353     return EAGAIN;
354 }
355
356 /* Tries to receive up to 'n' bytes from 'stream' into 'buffer', and returns:
357  *
358  *     - If successful, the number of bytes received (between 1 and 'n').
359  *
360  *     - On error, a negative errno value.
361  *
362  *     - 0, if the connection has been closed in the normal fashion, or if 'n'
363  *       is zero.
364  *
365  * The recv function will not block waiting for a packet to arrive.  If no
366  * data have been received, it returns -EAGAIN immediately. */
367 int
368 stream_recv(struct stream *stream, void *buffer, size_t n)
369 {
370     int retval = stream_connect(stream);
371     return (retval ? -retval
372             : n == 0 ? 0
373             : (stream->class->recv)(stream, buffer, n));
374 }
375
376 /* Tries to send up to 'n' bytes of 'buffer' on 'stream', and returns:
377  *
378  *     - If successful, the number of bytes sent (between 1 and 'n').  0 is
379  *       only a valid return value if 'n' is 0.
380  *
381  *     - On error, a negative errno value.
382  *
383  * The send function will not block.  If no bytes can be immediately accepted
384  * for transmission, it returns -EAGAIN immediately. */
385 int
386 stream_send(struct stream *stream, const void *buffer, size_t n)
387 {
388     int retval = stream_connect(stream);
389     return (retval ? -retval
390             : n == 0 ? 0
391             : (stream->class->send)(stream, buffer, n));
392 }
393
394 /* Allows 'stream' to perform maintenance activities, such as flushing
395  * output buffers. */
396 void
397 stream_run(struct stream *stream)
398 {
399     if (stream->class->run) {
400         (stream->class->run)(stream);
401     }
402 }
403
404 /* Arranges for the poll loop to wake up when 'stream' needs to perform
405  * maintenance activities. */
406 void
407 stream_run_wait(struct stream *stream)
408 {
409     if (stream->class->run_wait) {
410         (stream->class->run_wait)(stream);
411     }
412 }
413
414 /* Arranges for the poll loop to wake up when 'stream' is ready to take an
415  * action of the given 'type'. */
416 void
417 stream_wait(struct stream *stream, enum stream_wait_type wait)
418 {
419     assert(wait == STREAM_CONNECT || wait == STREAM_RECV
420            || wait == STREAM_SEND);
421
422     switch (stream->state) {
423     case SCS_CONNECTING:
424         wait = STREAM_CONNECT;
425         break;
426
427     case SCS_DISCONNECTED:
428         poll_immediate_wake();
429         return;
430     }
431     (stream->class->wait)(stream, wait);
432 }
433
434 void
435 stream_connect_wait(struct stream *stream)
436 {
437     stream_wait(stream, STREAM_CONNECT);
438 }
439
440 void
441 stream_recv_wait(struct stream *stream)
442 {
443     stream_wait(stream, STREAM_RECV);
444 }
445
446 void
447 stream_send_wait(struct stream *stream)
448 {
449     stream_wait(stream, STREAM_SEND);
450 }
451
452 /* Given 'name', a pstream name in the form "TYPE:ARGS", stores the class
453  * named "TYPE" into '*classp' and returns 0.  Returns EAFNOSUPPORT and stores
454  * a null pointer into '*classp' if 'name' is in the wrong form or if no such
455  * class exists. */
456 static int
457 pstream_lookup_class(const char *name, struct pstream_class **classp)
458 {
459     size_t prefix_len;
460     size_t i;
461
462     check_stream_classes();
463
464     *classp = NULL;
465     prefix_len = strcspn(name, ":");
466     if (name[prefix_len] == '\0') {
467         return EAFNOSUPPORT;
468     }
469     for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) {
470         struct pstream_class *class = pstream_classes[i];
471         if (strlen(class->name) == prefix_len
472             && !memcmp(class->name, name, prefix_len)) {
473             *classp = class;
474             return 0;
475         }
476     }
477     return EAFNOSUPPORT;
478 }
479
480 /* Returns 0 if 'name' is a pstream name in the form "TYPE:ARGS" and TYPE is
481  * a supported pstream type, otherwise EAFNOSUPPORT.  */
482 int
483 pstream_verify_name(const char *name)
484 {
485     struct pstream_class *class;
486     return pstream_lookup_class(name, &class);
487 }
488
489 /* Attempts to start listening for remote stream connections.  'name' is a
490  * connection name in the form "TYPE:ARGS", where TYPE is an passive stream
491  * class's name and ARGS are stream class-specific.
492  *
493  * Returns 0 if successful, otherwise a positive errno value.  If successful,
494  * stores a pointer to the new connection in '*pstreamp', otherwise a null
495  * pointer.  */
496 int
497 pstream_open(const char *name, struct pstream **pstreamp)
498 {
499     struct pstream_class *class;
500     struct pstream *pstream;
501     char *suffix_copy;
502     int error;
503
504     COVERAGE_INC(pstream_open);
505
506     /* Look up the class. */
507     error = pstream_lookup_class(name, &class);
508     if (!class) {
509         goto error;
510     }
511
512     /* Call class's "open" function. */
513     suffix_copy = xstrdup(strchr(name, ':') + 1);
514     error = class->listen(name, suffix_copy, &pstream);
515     free(suffix_copy);
516     if (error) {
517         goto error;
518     }
519
520     /* Success. */
521     *pstreamp = pstream;
522     return 0;
523
524 error:
525     *pstreamp = NULL;
526     return error;
527 }
528
529 /* Returns the name that was used to open 'pstream'.  The caller must not
530  * modify or free the name. */
531 const char *
532 pstream_get_name(const struct pstream *pstream)
533 {
534     return pstream->name;
535 }
536
537 /* Closes 'pstream'. */
538 void
539 pstream_close(struct pstream *pstream)
540 {
541     if (pstream != NULL) {
542         char *name = pstream->name;
543         (pstream->class->close)(pstream);
544         free(name);
545     }
546 }
547
548 /* Tries to accept a new connection on 'pstream'.  If successful, stores the
549  * new connection in '*new_stream' and returns 0.  Otherwise, returns a
550  * positive errno value.
551  *
552  * pstream_accept() will not block waiting for a connection.  If no connection
553  * is ready to be accepted, it returns EAGAIN immediately. */
554 int
555 pstream_accept(struct pstream *pstream, struct stream **new_stream)
556 {
557     int retval = (pstream->class->accept)(pstream, new_stream);
558     if (retval) {
559         *new_stream = NULL;
560     } else {
561         assert((*new_stream)->state != SCS_CONNECTING
562                || (*new_stream)->class->connect);
563     }
564     return retval;
565 }
566
567 /* Tries to accept a new connection on 'pstream'.  If successful, stores the
568  * new connection in '*new_stream' and returns 0.  Otherwise, returns a
569  * positive errno value.
570  *
571  * pstream_accept_block() blocks until a connection is ready or until an error
572  * occurs.  It will not return EAGAIN. */
573 int
574 pstream_accept_block(struct pstream *pstream, struct stream **new_stream)
575 {
576     int error;
577
578     fatal_signal_run();
579     while ((error = pstream_accept(pstream, new_stream)) == EAGAIN) {
580         pstream_wait(pstream);
581         poll_block();
582     }
583     if (error) {
584         *new_stream = NULL;
585     }
586     return error;
587 }
588
589 void
590 pstream_wait(struct pstream *pstream)
591 {
592     (pstream->class->wait)(pstream);
593 }
594 \f
595 /* Initializes 'stream' as a new stream named 'name', implemented via 'class'.
596  * The initial connection status, supplied as 'connect_status', is interpreted
597  * as follows:
598  *
599  *      - 0: 'stream' is connected.  Its 'send' and 'recv' functions may be
600  *        called in the normal fashion.
601  *
602  *      - EAGAIN: 'stream' is trying to complete a connection.  Its 'connect'
603  *        function should be called to complete the connection.
604  *
605  *      - Other positive errno values indicate that the connection failed with
606  *        the specified error.
607  *
608  * After calling this function, stream_close() must be used to destroy
609  * 'stream', otherwise resources will be leaked.
610  *
611  * The caller retains ownership of 'name'. */
612 void
613 stream_init(struct stream *stream, struct stream_class *class,
614             int connect_status, const char *name)
615 {
616     stream->class = class;
617     stream->state = (connect_status == EAGAIN ? SCS_CONNECTING
618                     : !connect_status ? SCS_CONNECTED
619                     : SCS_DISCONNECTED);
620     stream->error = connect_status;
621     stream->name = xstrdup(name);
622     assert(stream->state != SCS_CONNECTING || class->connect);
623 }
624
625 void
626 stream_set_remote_ip(struct stream *stream, uint32_t ip)
627 {
628     stream->remote_ip = ip;
629 }
630
631 void
632 stream_set_remote_port(struct stream *stream, uint16_t port)
633 {
634     stream->remote_port = port;
635 }
636
637 void
638 stream_set_local_ip(struct stream *stream, uint32_t ip)
639 {
640     stream->local_ip = ip;
641 }
642
643 void
644 stream_set_local_port(struct stream *stream, uint16_t port)
645 {
646     stream->local_port = port;
647 }
648
649 void
650 pstream_init(struct pstream *pstream, struct pstream_class *class,
651             const char *name)
652 {
653     pstream->class = class;
654     pstream->name = xstrdup(name);
655 }
656 \f
657 static int
658 count_fields(const char *s_)
659 {
660     char *s, *field, *save_ptr;
661     int n = 0;
662
663     save_ptr = NULL;
664     s = xstrdup(s_);
665     for (field = strtok_r(s, ":", &save_ptr); field != NULL;
666          field = strtok_r(NULL, ":", &save_ptr)) {
667         n++;
668     }
669     free(s);
670
671     return n;
672 }
673
674 /* Like stream_open(), but for tcp streams the port defaults to
675  * 'default_tcp_port' if no port number is given and for SSL streams the port
676  * defaults to 'default_ssl_port' if no port number is given. */
677 int
678 stream_open_with_default_ports(const char *name_,
679                                uint16_t default_tcp_port,
680                                uint16_t default_ssl_port,
681                                struct stream **streamp)
682 {
683     char *name;
684     int error;
685
686     if (!strncmp(name_, "tcp:", 4) && count_fields(name_) < 3) {
687         name = xasprintf("%s:%d", name_, default_tcp_port);
688     } else if (!strncmp(name_, "ssl:", 4) && count_fields(name_) < 3) {
689         name = xasprintf("%s:%d", name_, default_ssl_port);
690     } else {
691         name = xstrdup(name_);
692     }
693     error = stream_open(name, streamp);
694     free(name);
695
696     return error;
697 }
698
699 /* Like pstream_open(), but for ptcp streams the port defaults to
700  * 'default_ptcp_port' if no port number is given and for passive SSL streams
701  * the port defaults to 'default_pssl_port' if no port number is given. */
702 int
703 pstream_open_with_default_ports(const char *name_,
704                                 uint16_t default_ptcp_port,
705                                 uint16_t default_pssl_port,
706                                 struct pstream **pstreamp)
707 {
708     char *name;
709     int error;
710
711     if (!strncmp(name_, "ptcp:", 5) && count_fields(name_) < 2) {
712         name = xasprintf("%s%d", name_, default_ptcp_port);
713     } else if (!strncmp(name_, "pssl:", 5) && count_fields(name_) < 2) {
714         name = xasprintf("%s%d", name_, default_pssl_port);
715     } else {
716         name = xstrdup(name_);
717     }
718     error = pstream_open(name, pstreamp);
719     free(name);
720
721     return error;
722 }
723 \f
724 /* Attempts to guess the content type of a stream whose first few bytes were
725  * the 'size' bytes of 'data'. */
726 static enum stream_content_type
727 stream_guess_content(const uint8_t *data, size_t size)
728 {
729     if (size >= 2) {
730 #define PAIR(A, B) (((A) << 8) | (B))
731         switch (PAIR(data[0], data[1])) {
732         case PAIR(0x16, 0x03):  /* Handshake, version 3. */
733             return STREAM_SSL;
734         case PAIR('{', '"'):
735             return STREAM_JSONRPC;
736         case PAIR(OFP_VERSION, OFPT_HELLO):
737             return STREAM_OPENFLOW;
738         }
739     }
740
741     return STREAM_UNKNOWN;
742 }
743
744 /* Returns a string represenation of 'type'. */
745 static const char *
746 stream_content_type_to_string(enum stream_content_type type)
747 {
748     switch (type) {
749     case STREAM_UNKNOWN:
750     default:
751         return "unknown";
752
753     case STREAM_JSONRPC:
754         return "JSON-RPC";
755
756     case STREAM_OPENFLOW:
757         return "OpenFlow";
758
759     case STREAM_SSL:
760         return "SSL";
761     }
762 }
763
764 /* Attempts to guess the content type of a stream whose first few bytes were
765  * the 'size' bytes of 'data'.  If this is done successfully, and the guessed
766  * content type is other than 'expected_type', then log a message in vlog
767  * module 'module', naming 'stream_name' as the source, explaining what
768  * content was expected and what was actually received. */
769 void
770 stream_report_content(const void *data, size_t size,
771                       enum stream_content_type expected_type,
772                       struct vlog_module *module, const char *stream_name)
773 {
774     static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
775     enum stream_content_type actual_type;
776
777     actual_type = stream_guess_content(data, size);
778     if (actual_type != expected_type && actual_type != STREAM_UNKNOWN) {
779         vlog_rate_limit(module, VLL_WARN, &rl,
780                         "%s: received %s data on %s channel",
781                         stream_name,
782                         stream_content_type_to_string(expected_type),
783                         stream_content_type_to_string(actual_type));
784     }
785 }