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