lib/vconn-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 <assert.h>
  19 #include <errno.h>
  20 #include <poll.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <sys/types.h>
  24 #include <unistd.h>
  25 #include "fatal-signal.h"
  26 #include "leak-checker.h"
  27 #include "ofpbuf.h"
  28 #include "openflow/openflow.h"
  29 #include "poll-loop.h"
  30 #include "socket-util.h"
  31 #include "stream.h"
  32 #include "util.h"
  33 #include "vconn-provider.h"
  34 #include "vconn.h"
  35
  36 #include "vlog.h"
  37 #define THIS_MODULE VLM_vconn_stream
  38
  39 /* Active stream socket vconn. */
  40
  41 struct vconn_stream
  42 {
  43     struct vconn vconn;
  44     struct stream *stream;
  45     struct ofpbuf *rxbuf;
  46     struct ofpbuf *txbuf;
  47 };
  48
  49 static struct vconn_class stream_vconn_class;
  50
  51 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(10, 25);
  52
  53 static void vconn_stream_clear_txbuf(struct vconn_stream *);
  54 static int count_fields(const char *);
  55
  56 static struct vconn *
  57 vconn_stream_new(struct stream *stream, int connect_status)
  58 {
  59     struct vconn_stream *s;
  60
  61     s = xmalloc(sizeof *s);
  62     vconn_init(&s->vconn, &stream_vconn_class, connect_status,
  63                stream_get_name(stream));
  64     s->stream = stream;
  65     s->txbuf = NULL;
  66     s->rxbuf = NULL;
  67     return &s->vconn;
  68 }
  69
  70 /* Creates a new vconn that will send and receive data on a stream named 'name'
  71  * and stores a pointer to the vconn in '*vconnp'.
  72  *
  73  * Returns 0 if successful, otherwise a positive errno value. */
  74 static int
  75 vconn_stream_open(const char *name_, char *suffix UNUSED,
  76                   struct vconn **vconnp)
  77 {
  78     struct stream *stream;
  79     char *name;
  80     int error;
  81
  82     if (!strncmp(name_, "tcp:", 4) && count_fields(name_) < 3) {
  83         name = xasprintf("%s:%d", name_, OFP_TCP_PORT);
  84     } else if (!strncmp(name_, "ssl:", 4) && count_fields(name_) < 3) {
  85         name = xasprintf("%s:%d", name_, OFP_SSL_PORT);
  86     } else {
  87         name = xstrdup(name_);
  88     }
  89     error = stream_open(name, &stream);
  90     free(name);
  91
  92     if (error && error != EAGAIN) {
  93         return error;
  94     }
  95
  96     *vconnp = vconn_stream_new(stream, error);
  97     return 0;
  98 }
  99
 100 static struct vconn_stream *
 101 vconn_stream_cast(struct vconn *vconn)
 102 {
 103     return CONTAINER_OF(vconn, struct vconn_stream, vconn);
 104 }
 105
 106 static void
 107 vconn_stream_close(struct vconn *vconn)
 108 {
 109     struct vconn_stream *s = vconn_stream_cast(vconn);
 110     stream_close(s->stream);
 111     vconn_stream_clear_txbuf(s);
 112     ofpbuf_delete(s->rxbuf);
 113     free(s);
 114 }
 115
 116 static int
 117 vconn_stream_connect(struct vconn *vconn)
 118 {
 119     struct vconn_stream *s = vconn_stream_cast(vconn);
 120     return stream_connect(s->stream);
 121 }
 122
 123 static int
 124 vconn_stream_recv(struct vconn *vconn, struct ofpbuf **bufferp)
 125 {
 126     struct vconn_stream *s = vconn_stream_cast(vconn);
 127     struct ofpbuf *rx;
 128     size_t want_bytes;
 129     ssize_t retval;
 130
 131     if (s->rxbuf == NULL) {
 132         s->rxbuf = ofpbuf_new(1564);
 133     }
 134     rx = s->rxbuf;
 135
 136 again:
 137     if (sizeof(struct ofp_header) > rx->size) {
 138         want_bytes = sizeof(struct ofp_header) - rx->size;
 139     } else {
 140         struct ofp_header *oh = rx->data;
 141         size_t length = ntohs(oh->length);
 142         if (length < sizeof(struct ofp_header)) {
 143             VLOG_ERR_RL(&rl, "received too-short ofp_header (%zu bytes)",
 144                         length);
 145             return EPROTO;
 146         }
 147         want_bytes = length - rx->size;
 148         if (!want_bytes) {
 149             *bufferp = rx;
 150             s->rxbuf = NULL;
 151             return 0;
 152         }
 153     }
 154     ofpbuf_prealloc_tailroom(rx, want_bytes);
 155
 156     retval = stream_recv(s->stream, ofpbuf_tail(rx), want_bytes);
 157     if (retval > 0) {
 158         rx->size += retval;
 159         if (retval == want_bytes) {
 160             if (rx->size > sizeof(struct ofp_header)) {
 161                 *bufferp = rx;
 162                 s->rxbuf = NULL;
 163                 return 0;
 164             } else {
 165                 goto again;
 166             }
 167         }
 168         return EAGAIN;
 169     } else if (retval == 0) {
 170         if (rx->size) {
 171             VLOG_ERR_RL(&rl, "connection dropped mid-packet");
 172             return EPROTO;
 173         } else {
 174             return EOF;
 175         }
 176     } else {
 177         return -retval;
 178     }
 179 }
 180
 181 static void
 182 vconn_stream_clear_txbuf(struct vconn_stream *s)
 183 {
 184     ofpbuf_delete(s->txbuf);
 185     s->txbuf = NULL;
 186 }
 187
 188 static int
 189 vconn_stream_send(struct vconn *vconn, struct ofpbuf *buffer)
 190 {
 191     struct vconn_stream *s = vconn_stream_cast(vconn);
 192     ssize_t retval;
 193
 194     if (s->txbuf) {
 195         return EAGAIN;
 196     }
 197
 198     retval = stream_send(s->stream, buffer->data, buffer->size);
 199     if (retval == buffer->size) {
 200         ofpbuf_delete(buffer);
 201         return 0;
 202     } else if (retval >= 0 || retval == -EAGAIN) {
 203         leak_checker_claim(buffer);
 204         s->txbuf = buffer;
 205         if (retval > 0) {
 206             ofpbuf_pull(buffer, retval);
 207         }
 208         return 0;
 209     } else {
 210         return -retval;
 211     }
 212 }
 213
 214 static void
 215 vconn_stream_run(struct vconn *vconn)
 216 {
 217     struct vconn_stream *s = vconn_stream_cast(vconn);
 218     ssize_t retval;
 219
 220     if (!s->txbuf) {
 221         return;
 222     }
 223
 224     retval = stream_send(s->stream, s->txbuf->data, s->txbuf->size);
 225     if (retval < 0) {
 226         if (retval != -EAGAIN) {
 227             VLOG_ERR_RL(&rl, "send: %s", strerror(-retval));
 228             vconn_stream_clear_txbuf(s);
 229             return;
 230         }
 231     } else if (retval > 0) {
 232         ofpbuf_pull(s->txbuf, retval);
 233         if (!s->txbuf->size) {
 234             vconn_stream_clear_txbuf(s);
 235             return;
 236         }
 237     }
 238 }
 239
 240 static void
 241 vconn_stream_run_wait(struct vconn *vconn)
 242 {
 243     struct vconn_stream *s = vconn_stream_cast(vconn);
 244
 245     if (s->txbuf) {
 246         stream_send_wait(s->stream);
 247     }
 248 }
 249
 250 static void
 251 vconn_stream_wait(struct vconn *vconn, enum vconn_wait_type wait)
 252 {
 253     struct vconn_stream *s = vconn_stream_cast(vconn);
 254     switch (wait) {
 255     case WAIT_CONNECT:
 256         stream_connect_wait(s->stream);
 257         break;
 258
 259     case WAIT_SEND:
 260         if (!s->txbuf) {
 261             stream_send_wait(s->stream);
 262         } else {
 263             /* Nothing to do: need to drain txbuf first.
 264              * vconn_stream_run_wait() will arrange to wake up when there room
 265              * to send data, so there's no point in calling poll_fd_wait()
 266              * redundantly here. */
 267         }
 268         break;
 269
 270     case WAIT_RECV:
 271         stream_recv_wait(s->stream);
 272         break;
 273
 274     default:
 275         NOT_REACHED();
 276     }
 277 }
 278 \f
 279 /* Passive stream socket vconn. */
 280
 281 struct pvconn_pstream
 282 {
 283     struct pvconn pvconn;
 284     struct pstream *pstream;
 285 };
 286
 287 static struct pvconn_class pstream_pvconn_class;
 288
 289 static struct pvconn_pstream *
 290 pvconn_pstream_cast(struct pvconn *pvconn)
 291 {
 292     return CONTAINER_OF(pvconn, struct pvconn_pstream, pvconn);
 293 }
 294
 295 /* Creates a new pvconn named 'name' that will accept new connections using
 296  * pstream_accept() and stores a pointer to the pvconn in '*pvconnp'.
 297  *
 298  * Returns 0 if successful, otherwise a positive errno value.  (The current
 299  * implementation never fails.) */
 300 static int
 301 pvconn_pstream_listen(const char *name_, char *suffix UNUSED,
 302                       struct pvconn **pvconnp)
 303 {
 304     struct pvconn_pstream *ps;
 305     struct pstream *pstream;
 306     char *name;
 307     int error;
 308
 309     if (!strncmp(name_, "ptcp:", 5) && count_fields(name_) < 2) {
 310         name = xasprintf("%s:%d", name_, OFP_TCP_PORT);
 311     } else if (!strncmp(name_, "pssl:", 5) && count_fields(name_) < 2) {
 312         name = xasprintf("%s:%d", name_, OFP_SSL_PORT);
 313     } else {
 314         name = xstrdup(name_);
 315     }
 316     error = pstream_open(name, &pstream);
 317     free(name);
 318     if (error) {
 319         return error;
 320     }
 321
 322     ps = xmalloc(sizeof *ps);
 323     pvconn_init(&ps->pvconn, &pstream_pvconn_class, name_);
 324     ps->pstream = pstream;
 325     *pvconnp = &ps->pvconn;
 326     return 0;
 327 }
 328
 329 static void
 330 pvconn_pstream_close(struct pvconn *pvconn)
 331 {
 332     struct pvconn_pstream *ps = pvconn_pstream_cast(pvconn);
 333     pstream_close(ps->pstream);
 334     free(ps);
 335 }
 336
 337 static int
 338 pvconn_pstream_accept(struct pvconn *pvconn, struct vconn **new_vconnp)
 339 {
 340     struct pvconn_pstream *ps = pvconn_pstream_cast(pvconn);
 341     struct stream *stream;
 342     int error;
 343
 344     error = pstream_accept(ps->pstream, &stream);
 345     if (error) {
 346         if (error != EAGAIN) {
 347             VLOG_DBG_RL(&rl, "%s: accept: %s",
 348                         pstream_get_name(ps->pstream), strerror(error));
 349         }
 350         return error;
 351     }
 352
 353     *new_vconnp = vconn_stream_new(stream, 0);
 354     return 0;
 355 }
 356
 357 static void
 358 pvconn_pstream_wait(struct pvconn *pvconn)
 359 {
 360     struct pvconn_pstream *ps = pvconn_pstream_cast(pvconn);
 361     pstream_wait(ps->pstream);
 362 }
 363 \f
 364 static int
 365 count_fields(const char *s_)
 366 {
 367     char *s, *field, *save_ptr;
 368     int n = 0;
 369
 370     save_ptr = NULL;
 371     s = xstrdup(s_);
 372     for (field = strtok_r(s, ":", &save_ptr); field != NULL;
 373          field = strtok_r(NULL, ":", &save_ptr)) {
 374         n++;
 375     }
 376     free(s);
 377
 378     return n;
 379 }
 380 \f
 381 /* Stream-based vconns and pvconns. */
 382
 383 #define DEFINE_VCONN_STREAM_CLASS(NAME)             \
 384         struct vconn_class NAME##_vconn_class = {   \
 385             #NAME,                                  \
 386             vconn_stream_open,                      \
 387             vconn_stream_close,                     \
 388             vconn_stream_connect,                   \
 389             vconn_stream_recv,                      \
 390             vconn_stream_send,                      \
 391             vconn_stream_run,                       \
 392             vconn_stream_run_wait,                  \
 393             vconn_stream_wait,                      \
 394         };
 395
 396 #define DEFINE_PVCONN_STREAM_CLASS(NAME)            \
 397         struct pvconn_class NAME##_pvconn_class = { \
 398             #NAME,                                  \
 399             pvconn_pstream_listen,                  \
 400             pvconn_pstream_close,                   \
 401             pvconn_pstream_accept,                  \
 402             pvconn_pstream_wait                     \
 403         };
 404
 405 static DEFINE_VCONN_STREAM_CLASS(stream);
 406 static DEFINE_PVCONN_STREAM_CLASS(pstream);
 407
 408 DEFINE_VCONN_STREAM_CLASS(tcp);
 409 DEFINE_PVCONN_STREAM_CLASS(ptcp);
 410
 411 DEFINE_VCONN_STREAM_CLASS(unix);
 412 DEFINE_PVCONN_STREAM_CLASS(punix);
 413
 414 #ifdef HAVE_OPENSSL
 415 DEFINE_VCONN_STREAM_CLASS(ssl);
 416 DEFINE_PVCONN_STREAM_CLASS(pssl);
 417 #endif