lib/stream.c

   1 /*
   2  * Copyright (c) 2008, 2009 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 "flow.h"
  29 #include "ofp-print.h"
  30 #include "ofpbuf.h"
  31 #include "openflow/nicira-ext.h"
  32 #include "openflow/openflow.h"
  33 #include "packets.h"
  34 #include "poll-loop.h"
  35 #include "random.h"
  36 #include "util.h"
  37
  38 #define THIS_MODULE VLM_stream
  39 #include "vlog.h"
  40
  41 /* State of an active stream.*/
  42 enum stream_state {
  43     SCS_CONNECTING,             /* Underlying stream is not connected. */
  44     SCS_CONNECTED,              /* Connection established. */
  45     SCS_DISCONNECTED            /* Connection failed or connection closed. */
  46 };
  47
  48 static struct stream_class *stream_classes[] = {
  49     &tcp_stream_class,
  50     &unix_stream_class,
  51 };
  52
  53 static struct pstream_class *pstream_classes[] = {
  54     &ptcp_pstream_class,
  55     &punix_pstream_class,
  56 };
  57
  58 /* Check the validity of the stream class structures. */
  59 static void
  60 check_stream_classes(void)
  61 {
  62 #ifndef NDEBUG
  63     size_t i;
  64
  65     for (i = 0; i < ARRAY_SIZE(stream_classes); i++) {
  66         struct stream_class *class = stream_classes[i];
  67         assert(class->name != NULL);
  68         assert(class->open != NULL);
  69         if (class->close || class->recv || class->send || class->wait) {
  70             assert(class->close != NULL);
  71             assert(class->recv != NULL);
  72             assert(class->send != NULL);
  73             assert(class->wait != NULL);
  74         } else {
  75             /* This class delegates to another one. */
  76         }
  77     }
  78
  79     for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) {
  80         struct pstream_class *class = pstream_classes[i];
  81         assert(class->name != NULL);
  82         assert(class->listen != NULL);
  83         if (class->close || class->accept || class->wait) {
  84             assert(class->close != NULL);
  85             assert(class->accept != NULL);
  86             assert(class->wait != NULL);
  87         } else {
  88             /* This class delegates to another one. */
  89         }
  90     }
  91 #endif
  92 }
  93
  94 /* Prints information on active (if 'active') and passive (if 'passive')
  95  * connection methods supported by the stream. */
  96 void
  97 stream_usage(const char *name, bool active, bool passive)
  98 {
  99     /* Really this should be implemented via callbacks into the stream
 100      * providers, but that seems too heavy-weight to bother with at the
 101      * moment. */
 102
 103     printf("\n");
 104     if (active) {
 105         printf("Active %s connection methods:\n", name);
 106         printf("  tcp:IP:PORT             "
 107                "PORT at remote IP\n");
 108         printf("  unix:FILE               "
 109                "Unix domain socket named FILE\n");
 110     }
 111
 112     if (passive) {
 113         printf("Passive %s connection methods:\n", name);
 114         printf("  ptcp:PORT[:IP]          "
 115                "listen to TCP PORT on IP\n");
 116         printf("  punix:FILE              "
 117                "listen on Unix domain socket FILE\n");
 118     }
 119 }
 120
 121 /* Attempts to connect a stream to a remote peer.  'name' is a connection name
 122  * in the form "TYPE:ARGS", where TYPE is an active stream class's name and
 123  * ARGS are stream class-specific.
 124  *
 125  * Returns 0 if successful, otherwise a positive errno value.  If successful,
 126  * stores a pointer to the new connection in '*streamp', otherwise a null
 127  * pointer.  */
 128 int
 129 stream_open(const char *name, struct stream **streamp)
 130 {
 131     size_t prefix_len;
 132     size_t i;
 133
 134     COVERAGE_INC(stream_open);
 135     check_stream_classes();
 136
 137     *streamp = NULL;
 138     prefix_len = strcspn(name, ":");
 139     if (prefix_len == strlen(name)) {
 140         return EAFNOSUPPORT;
 141     }
 142     for (i = 0; i < ARRAY_SIZE(stream_classes); i++) {
 143         struct stream_class *class = stream_classes[i];
 144         if (strlen(class->name) == prefix_len
 145             && !memcmp(class->name, name, prefix_len)) {
 146             struct stream *stream;
 147             char *suffix_copy = xstrdup(name + prefix_len + 1);
 148             int retval = class->open(name, suffix_copy, &stream);
 149             free(suffix_copy);
 150             if (!retval) {
 151                 assert(stream->state != SCS_CONNECTING
 152                        || stream->class->connect);
 153                 *streamp = stream;
 154             }
 155             return retval;
 156         }
 157     }
 158     return EAFNOSUPPORT;
 159 }
 160
 161 int
 162 stream_open_block(const char *name, struct stream **streamp)
 163 {
 164     struct stream *stream;
 165     int error;
 166
 167     error = stream_open(name, &stream);
 168     while (error == EAGAIN) {
 169         stream_connect_wait(stream);
 170         poll_block();
 171         error = stream_connect(stream);
 172         assert(error != EINPROGRESS);
 173     }
 174     if (error) {
 175         stream_close(stream);
 176         *streamp = NULL;
 177     } else {
 178         *streamp = stream;
 179     }
 180     return error;
 181 }
 182
 183 /* Closes 'stream'. */
 184 void
 185 stream_close(struct stream *stream)
 186 {
 187     if (stream != NULL) {
 188         char *name = stream->name;
 189         (stream->class->close)(stream);
 190         free(name);
 191     }
 192 }
 193
 194 /* Returns the name of 'stream', that is, the string passed to
 195  * stream_open(). */
 196 const char *
 197 stream_get_name(const struct stream *stream)
 198 {
 199     return stream ? stream->name : "(null)";
 200 }
 201
 202 /* Returns the IP address of the peer, or 0 if the peer is not connected over
 203  * an IP-based protocol or if its IP address is not yet known. */
 204 uint32_t
 205 stream_get_remote_ip(const struct stream *stream)
 206 {
 207     return stream->remote_ip;
 208 }
 209
 210 /* Returns the transport port of the peer, or 0 if the connection does not
 211  * contain a port or if the port is not yet known. */
 212 uint16_t
 213 stream_get_remote_port(const struct stream *stream)
 214 {
 215     return stream->remote_port;
 216 }
 217
 218 /* Returns the IP address used to connect to the peer, or 0 if the connection
 219  * is not an IP-based protocol or if its IP address is not yet known. */
 220 uint32_t
 221 stream_get_local_ip(const struct stream *stream)
 222 {
 223     return stream->local_ip;
 224 }
 225
 226 /* Returns the transport port used to connect to the peer, or 0 if the
 227  * connection does not contain a port or if the port is not yet known. */
 228 uint16_t
 229 stream_get_local_port(const struct stream *stream)
 230 {
 231     return stream->local_port;
 232 }
 233
 234 static void
 235 scs_connecting(struct stream *stream)
 236 {
 237     int retval = (stream->class->connect)(stream);
 238     assert(retval != EINPROGRESS);
 239     if (!retval) {
 240         stream->state = SCS_CONNECTED;
 241     } else if (retval != EAGAIN) {
 242         stream->state = SCS_DISCONNECTED;
 243         stream->error = retval;
 244     }
 245 }
 246
 247 /* Tries to complete the connection on 'stream', which must be an active
 248  * stream.  If 'stream''s connection is complete, returns 0 if the connection
 249  * was successful or a positive errno value if it failed.  If the
 250  * connection is still in progress, returns EAGAIN. */
 251 int
 252 stream_connect(struct stream *stream)
 253 {
 254     enum stream_state last_state;
 255
 256     do {
 257         last_state = stream->state;
 258         switch (stream->state) {
 259         case SCS_CONNECTING:
 260             scs_connecting(stream);
 261             break;
 262
 263         case SCS_CONNECTED:
 264             return 0;
 265
 266         case SCS_DISCONNECTED:
 267             return stream->error;
 268
 269         default:
 270             NOT_REACHED();
 271         }
 272     } while (stream->state != last_state);
 273
 274     return EAGAIN;
 275 }
 276
 277 /* Tries to receive up to 'n' bytes from 'stream' into 'buffer', and returns:
 278  *
 279  *     - If successful, the number of bytes received (between 1 and 'n').
 280  *
 281  *     - On error, a negative errno value.
 282  *
 283  *     - 0, if the connection has been closed in the normal fashion, or if 'n'
 284  *       is zero.
 285  *
 286  * The recv function will not block waiting for a packet to arrive.  If no
 287  * data have been received, it returns -EAGAIN immediately. */
 288 int
 289 stream_recv(struct stream *stream, void *buffer, size_t n)
 290 {
 291     int retval = stream_connect(stream);
 292     return (retval ? -retval
 293             : n == 0 ? 0
 294             : (stream->class->recv)(stream, buffer, n));
 295 }
 296
 297 /* Tries to send up to 'n' bytes of 'buffer' on 'stream', and returns:
 298  *
 299  *     - If successful, the number of bytes sent (between 1 and 'n').  0 is
 300  *       only a valid return value if 'n' is 0.
 301  *
 302  *     - On error, a negative errno value.
 303  *
 304  * The send function will not block.  If no bytes can be immediately accepted
 305  * for transmission, it returns -EAGAIN immediately. */
 306 int
 307 stream_send(struct stream *stream, const void *buffer, size_t n)
 308 {
 309     int retval = stream_connect(stream);
 310     return (retval ? -retval
 311             : n == 0 ? 0
 312             : (stream->class->send)(stream, buffer, n));
 313 }
 314
 315 void
 316 stream_wait(struct stream *stream, enum stream_wait_type wait)
 317 {
 318     assert(wait == STREAM_CONNECT || wait == STREAM_RECV
 319            || wait == STREAM_SEND);
 320
 321     switch (stream->state) {
 322     case SCS_CONNECTING:
 323         wait = STREAM_CONNECT;
 324         break;
 325
 326     case SCS_DISCONNECTED:
 327         poll_immediate_wake();
 328         return;
 329     }
 330     (stream->class->wait)(stream, wait);
 331 }
 332
 333 void
 334 stream_connect_wait(struct stream *stream)
 335 {
 336     stream_wait(stream, STREAM_CONNECT);
 337 }
 338
 339 void
 340 stream_recv_wait(struct stream *stream)
 341 {
 342     stream_wait(stream, STREAM_RECV);
 343 }
 344
 345 void
 346 stream_send_wait(struct stream *stream)
 347 {
 348     stream_wait(stream, STREAM_SEND);
 349 }
 350
 351 /* Attempts to start listening for remote stream connections.  'name' is a
 352  * connection name in the form "TYPE:ARGS", where TYPE is an passive stream
 353  * class's name and ARGS are stream class-specific.
 354  *
 355  * Returns 0 if successful, otherwise a positive errno value.  If successful,
 356  * stores a pointer to the new connection in '*pstreamp', otherwise a null
 357  * pointer.  */
 358 int
 359 pstream_open(const char *name, struct pstream **pstreamp)
 360 {
 361     size_t prefix_len;
 362     size_t i;
 363
 364     check_stream_classes();
 365
 366     *pstreamp = NULL;
 367     prefix_len = strcspn(name, ":");
 368     if (prefix_len == strlen(name)) {
 369         return EAFNOSUPPORT;
 370     }
 371     for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) {
 372         struct pstream_class *class = pstream_classes[i];
 373         if (strlen(class->name) == prefix_len
 374             && !memcmp(class->name, name, prefix_len)) {
 375             char *suffix_copy = xstrdup(name + prefix_len + 1);
 376             int retval = class->listen(name, suffix_copy, pstreamp);
 377             free(suffix_copy);
 378             if (retval) {
 379                 *pstreamp = NULL;
 380             }
 381             return retval;
 382         }
 383     }
 384     return EAFNOSUPPORT;
 385 }
 386
 387 /* Returns the name that was used to open 'pstream'.  The caller must not
 388  * modify or free the name. */
 389 const char *
 390 pstream_get_name(const struct pstream *pstream)
 391 {
 392     return pstream->name;
 393 }
 394
 395 /* Closes 'pstream'. */
 396 void
 397 pstream_close(struct pstream *pstream)
 398 {
 399     if (pstream != NULL) {
 400         char *name = pstream->name;
 401         (pstream->class->close)(pstream);
 402         free(name);
 403     }
 404 }
 405
 406 /* Tries to accept a new connection on 'pstream'.  If successful, stores the
 407  * new connection in '*new_stream' and returns 0.  Otherwise, returns a
 408  * positive errno value.
 409  *
 410  * pstream_accept() will not block waiting for a connection.  If no connection
 411  * is ready to be accepted, it returns EAGAIN immediately. */
 412 int
 413 pstream_accept(struct pstream *pstream, struct stream **new_stream)
 414 {
 415     int retval = (pstream->class->accept)(pstream, new_stream);
 416     if (retval) {
 417         *new_stream = NULL;
 418     } else {
 419         assert((*new_stream)->state != SCS_CONNECTING
 420                || (*new_stream)->class->connect);
 421     }
 422     return retval;
 423 }
 424
 425 /* Tries to accept a new connection on 'pstream'.  If successful, stores the
 426  * new connection in '*new_stream' and returns 0.  Otherwise, returns a
 427  * positive errno value.
 428  *
 429  * pstream_accept_block() blocks until a connection is ready or until an error
 430  * occurs.  It will not return EAGAIN. */
 431 int
 432 pstream_accept_block(struct pstream *pstream, struct stream **new_stream)
 433 {
 434     int error;
 435
 436     while ((error = pstream_accept(pstream, new_stream)) == EAGAIN) {
 437         pstream_wait(pstream);
 438         poll_block();
 439     }
 440     if (error) {
 441         *new_stream = NULL;
 442     }
 443     return error;
 444 }
 445
 446 void
 447 pstream_wait(struct pstream *pstream)
 448 {
 449     (pstream->class->wait)(pstream);
 450 }
 451 \f
 452 /* Initializes 'stream' as a new stream named 'name', implemented via 'class'.
 453  * The initial connection status, supplied as 'connect_status', is interpreted
 454  * as follows:
 455  *
 456  *      - 0: 'stream' is connected.  Its 'send' and 'recv' functions may be
 457  *        called in the normal fashion.
 458  *
 459  *      - EAGAIN: 'stream' is trying to complete a connection.  Its 'connect'
 460  *        function should be called to complete the connection.
 461  *
 462  *      - Other positive errno values indicate that the connection failed with
 463  *        the specified error.
 464  *
 465  * After calling this function, stream_close() must be used to destroy
 466  * 'stream', otherwise resources will be leaked.
 467  *
 468  * The caller retains ownership of 'name'. */
 469 void
 470 stream_init(struct stream *stream, struct stream_class *class,
 471             int connect_status, const char *name)
 472 {
 473     stream->class = class;
 474     stream->state = (connect_status == EAGAIN ? SCS_CONNECTING
 475                     : !connect_status ? SCS_CONNECTED
 476                     : SCS_DISCONNECTED);
 477     stream->error = connect_status;
 478     stream->name = xstrdup(name);
 479 }
 480
 481 void
 482 stream_set_remote_ip(struct stream *stream, uint32_t ip)
 483 {
 484     stream->remote_ip = ip;
 485 }
 486
 487 void
 488 stream_set_remote_port(struct stream *stream, uint16_t port)
 489 {
 490     stream->remote_port = port;
 491 }
 492
 493 void
 494 stream_set_local_ip(struct stream *stream, uint32_t ip)
 495 {
 496     stream->local_ip = ip;
 497 }
 498
 499 void
 500 stream_set_local_port(struct stream *stream, uint16_t port)
 501 {
 502     stream->local_port = port;
 503 }
 504
 505 void
 506 pstream_init(struct pstream *pstream, struct pstream_class *class,
 507             const char *name)
 508 {
 509     pstream->class = class;
 510     pstream->name = xstrdup(name);
 511 }