AUTHORS: Add Valient Gough.
[openvswitch] / 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 }