lib/stream-fd.c

   1 /*
   2  * Copyright (c) 2008, 2009, 2010, 2012 Nicira, Inc.
   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-fd.h"
  19 #include <errno.h>
  20 #include <poll.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <sys/socket.h>
  24 #include <sys/types.h>
  25 #include <unistd.h>
  26 #include "fatal-signal.h"
  27 #include "poll-loop.h"
  28 #include "socket-util.h"
  29 #include "stress.h"
  30 #include "util.h"
  31 #include "stream-provider.h"
  32 #include "stream.h"
  33 #include "vlog.h"
  34
  35 VLOG_DEFINE_THIS_MODULE(stream_fd);
  36
  37 /* Active file descriptor stream. */
  38
  39 struct stream_fd
  40 {
  41     struct stream stream;
  42     int fd;
  43 };
  44
  45 static const struct stream_class stream_fd_class;
  46
  47 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(10, 25);
  48
  49 static void maybe_unlink_and_free(char *path);
  50
  51 /* Creates a new stream named 'name' that will send and receive data on 'fd'
  52  * and stores a pointer to the stream in '*streamp'.  Initial connection status
  53  * 'connect_status' is interpreted as described for stream_init().
  54  *
  55  * Returns 0 if successful, otherwise a positive errno value.  (The current
  56  * implementation never fails.) */
  57 int
  58 new_fd_stream(const char *name, int fd, int connect_status,
  59               struct stream **streamp)
  60 {
  61     struct stream_fd *s;
  62
  63     s = xmalloc(sizeof *s);
  64     stream_init(&s->stream, &stream_fd_class, connect_status, name);
  65     s->fd = fd;
  66     *streamp = &s->stream;
  67     return 0;
  68 }
  69
  70 static struct stream_fd *
  71 stream_fd_cast(struct stream *stream)
  72 {
  73     stream_assert_class(stream, &stream_fd_class);
  74     return CONTAINER_OF(stream, struct stream_fd, stream);
  75 }
  76
  77 static void
  78 fd_close(struct stream *stream)
  79 {
  80     struct stream_fd *s = stream_fd_cast(stream);
  81     close(s->fd);
  82     free(s);
  83 }
  84
  85 static int
  86 fd_connect(struct stream *stream)
  87 {
  88     struct stream_fd *s = stream_fd_cast(stream);
  89     return check_connection_completion(s->fd);
  90 }
  91
  92 STRESS_OPTION(
  93     stream_flaky_recv, "simulate failure of fd stream recvs",
  94     100, 0, -1, 0);
  95
  96 static ssize_t
  97 fd_recv(struct stream *stream, void *buffer, size_t n)
  98 {
  99     struct stream_fd *s = stream_fd_cast(stream);
 100     ssize_t retval;
 101
 102     if (STRESS(stream_flaky_recv)) {
 103         return -EIO;
 104     }
 105
 106     retval = read(s->fd, buffer, n);
 107     return retval >= 0 ? retval : -errno;
 108 }
 109
 110 STRESS_OPTION(
 111     stream_flaky_send, "simulate failure of fd stream sends",
 112     100, 0, -1, 0);
 113
 114 static ssize_t
 115 fd_send(struct stream *stream, const void *buffer, size_t n)
 116 {
 117     struct stream_fd *s = stream_fd_cast(stream);
 118     ssize_t retval;
 119
 120     if (STRESS(stream_flaky_send)) {
 121         return -EIO;
 122     }
 123
 124     retval = write(s->fd, buffer, n);
 125     return (retval > 0 ? retval
 126             : retval == 0 ? -EAGAIN
 127             : -errno);
 128 }
 129
 130 static void
 131 fd_wait(struct stream *stream, enum stream_wait_type wait)
 132 {
 133     struct stream_fd *s = stream_fd_cast(stream);
 134     switch (wait) {
 135     case STREAM_CONNECT:
 136     case STREAM_SEND:
 137         poll_fd_wait(s->fd, POLLOUT);
 138         break;
 139
 140     case STREAM_RECV:
 141         poll_fd_wait(s->fd, POLLIN);
 142         break;
 143
 144     default:
 145         NOT_REACHED();
 146     }
 147 }
 148
 149 static const struct stream_class stream_fd_class = {
 150     "fd",                       /* name */
 151     false,                      /* needs_probes */
 152     NULL,                       /* open */
 153     fd_close,                   /* close */
 154     fd_connect,                 /* connect */
 155     fd_recv,                    /* recv */
 156     fd_send,                    /* send */
 157     NULL,                       /* run */
 158     NULL,                       /* run_wait */
 159     fd_wait,                    /* wait */
 160 };
 161 \f
 162 /* Passive file descriptor stream. */
 163
 164 struct fd_pstream
 165 {
 166     struct pstream pstream;
 167     int fd;
 168     int (*accept_cb)(int fd, const struct sockaddr *, size_t sa_len,
 169                      struct stream **);
 170     int (*set_dscp_cb)(int fd, uint8_t dscp);
 171     char *unlink_path;
 172 };
 173
 174 static const struct pstream_class fd_pstream_class;
 175
 176 static struct fd_pstream *
 177 fd_pstream_cast(struct pstream *pstream)
 178 {
 179     pstream_assert_class(pstream, &fd_pstream_class);
 180     return CONTAINER_OF(pstream, struct fd_pstream, pstream);
 181 }
 182
 183 /* Creates a new pstream named 'name' that will accept new socket connections
 184  * on 'fd' and stores a pointer to the stream in '*pstreamp'.
 185  *
 186  * When a connection has been accepted, 'accept_cb' will be called with the new
 187  * socket fd 'fd' and the remote address of the connection 'sa' and 'sa_len'.
 188  * accept_cb must return 0 if the connection is successful, in which case it
 189  * must initialize '*streamp' to the new stream, or a positive errno value on
 190  * error.  In either case accept_cb takes ownership of the 'fd' passed in.
 191  *
 192  * When '*pstreamp' is closed, then 'unlink_path' (if nonnull) will be passed
 193  * to fatal_signal_unlink_file_now() and freed with free().
 194  *
 195  * Returns 0 if successful, otherwise a positive errno value.  (The current
 196  * implementation never fails.) */
 197 int
 198 new_fd_pstream(const char *name, int fd,
 199                int (*accept_cb)(int fd, const struct sockaddr *sa,
 200                                 size_t sa_len, struct stream **streamp),
 201                int (*set_dscp_cb)(int fd, uint8_t dscp),
 202                char *unlink_path, struct pstream **pstreamp)
 203 {
 204     struct fd_pstream *ps = xmalloc(sizeof *ps);
 205     pstream_init(&ps->pstream, &fd_pstream_class, name);
 206     ps->fd = fd;
 207     ps->accept_cb = accept_cb;
 208     ps->set_dscp_cb = set_dscp_cb;
 209     ps->unlink_path = unlink_path;
 210     *pstreamp = &ps->pstream;
 211     return 0;
 212 }
 213
 214 static void
 215 pfd_close(struct pstream *pstream)
 216 {
 217     struct fd_pstream *ps = fd_pstream_cast(pstream);
 218     close(ps->fd);
 219     maybe_unlink_and_free(ps->unlink_path);
 220     free(ps);
 221 }
 222
 223 static int
 224 pfd_accept(struct pstream *pstream, struct stream **new_streamp)
 225 {
 226     struct fd_pstream *ps = fd_pstream_cast(pstream);
 227     struct sockaddr_storage ss;
 228     socklen_t ss_len = sizeof ss;
 229     int new_fd;
 230     int retval;
 231
 232     new_fd = accept(ps->fd, (struct sockaddr *) &ss, &ss_len);
 233     if (new_fd < 0) {
 234         retval = errno;
 235         if (retval != EAGAIN) {
 236             VLOG_DBG_RL(&rl, "accept: %s", strerror(retval));
 237         }
 238         return retval;
 239     }
 240
 241     retval = set_nonblocking(new_fd);
 242     if (retval) {
 243         close(new_fd);
 244         return retval;
 245     }
 246
 247     return ps->accept_cb(new_fd, (const struct sockaddr *) &ss, ss_len,
 248                          new_streamp);
 249 }
 250
 251 static void
 252 pfd_wait(struct pstream *pstream)
 253 {
 254     struct fd_pstream *ps = fd_pstream_cast(pstream);
 255     poll_fd_wait(ps->fd, POLLIN);
 256 }
 257
 258 static int
 259 pfd_set_dscp(struct pstream *pstream, uint8_t dscp)
 260 {
 261     struct fd_pstream *ps = fd_pstream_cast(pstream);
 262     if (ps->set_dscp_cb) {
 263         return ps->set_dscp_cb(ps->fd, dscp);
 264     }
 265     return 0;
 266 }
 267
 268 static const struct pstream_class fd_pstream_class = {
 269     "pstream",
 270     false,
 271     NULL,
 272     pfd_close,
 273     pfd_accept,
 274     pfd_wait,
 275     pfd_set_dscp,
 276 };
 277 \f
 278 /* Helper functions. */
 279 static void
 280 maybe_unlink_and_free(char *path)
 281 {
 282     if (path) {
 283         fatal_signal_unlink_file_now(path);
 284         free(path);
 285     }
 286 }