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