Revert "use proper tun_ip definitions"
[sliver-openvswitch.git] / lib / worker.c
1 /* Copyright (c) 2012 Nicira, Inc.
2  *
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at:
6  *
7  *     http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15
16 #include <config.h>
17
18 #include "worker.h"
19
20 #include <assert.h>
21 #include <errno.h>
22 #include <stdlib.h>
23 #include <string.h>
24 #include <sys/socket.h>
25 #include <sys/types.h>
26 #include <sys/uio.h>
27 #include <sys/wait.h>
28 #include <unistd.h>
29
30 #include "command-line.h"
31 #include "daemon.h"
32 #include "ofpbuf.h"
33 #include "poll-loop.h"
34 #include "socket-util.h"
35 #include "util.h"
36 #include "vlog.h"
37
38 VLOG_DEFINE_THIS_MODULE(worker);
39
40 /* Header for an RPC request. */
41 struct worker_request {
42     size_t request_len;              /* Length of the payload in bytes. */
43     worker_request_func *request_cb; /* Function to call in worker process. */
44     worker_reply_func *reply_cb;     /* Function to call in main process. */
45     void *reply_aux;                 /* Auxiliary data for 'reply_cb'. */
46 };
47
48 /* Header for an RPC reply. */
49 struct worker_reply {
50     size_t reply_len;            /* Length of the payload in bytes. */
51     worker_reply_func *reply_cb; /* Function to call in main process. */
52     void *reply_aux;             /* Auxiliary data for 'reply_cb'. */
53 };
54
55 /* Receive buffer for a RPC request or reply. */
56 struct rxbuf {
57     /* Header. */
58     struct ofpbuf header;       /* Header data. */
59     int fds[SOUTIL_MAX_FDS];    /* File descriptors. */
60     size_t n_fds;
61
62     /* Payload. */
63     struct ofpbuf payload;      /* Payload data. */
64 };
65
66 static int client_sock = -1;
67 static struct rxbuf client_rx;
68
69 static void rxbuf_init(struct rxbuf *);
70 static void rxbuf_clear(struct rxbuf *);
71 static int rxbuf_run(struct rxbuf *, int sock, size_t header_len);
72
73 static struct iovec *prefix_iov(void *data, size_t len,
74                                 const struct iovec *iovs, size_t n_iovs);
75
76 static void worker_broke(void);
77
78 static void worker_main(int fd) NO_RETURN;
79
80 /* Starts a worker process as a subprocess of the current process.  Currently
81  * only a single worker process is supported, so this function may only be
82  * called once.
83  *
84  * The client should call worker_run() and worker_wait() from its main loop.
85  *
86  * Call this function between daemonize_start() and daemonize_complete(). */
87 void
88 worker_start(void)
89 {
90     int work_fds[2];
91
92     assert(client_sock < 0);
93
94     /* Create non-blocking socket pair. */
95     xsocketpair(AF_UNIX, SOCK_STREAM, 0, work_fds);
96     xset_nonblocking(work_fds[0]);
97     xset_nonblocking(work_fds[1]);
98
99     if (!fork_and_clean_up()) {
100         /* In child (worker) process. */
101         daemonize_post_detach();
102         close(work_fds[0]);
103         worker_main(work_fds[1]);
104         NOT_REACHED();
105     }
106
107     /* In parent (main) process. */
108     close(work_fds[1]);
109     client_sock = work_fds[0];
110     rxbuf_init(&client_rx);
111 }
112
113 /* Returns true if this process has started a worker and the worker is not
114  * known to have malfunctioned. */
115 bool
116 worker_is_running(void)
117 {
118     return client_sock >= 0;
119 }
120
121 /* If a worker process was started, processes RPC replies from it, calling the
122  * registered 'reply_cb' callbacks.
123  *
124  * If the worker process died or malfunctioned, aborts. */
125 void
126 worker_run(void)
127 {
128     if (worker_is_running()) {
129         int error;
130
131         error = rxbuf_run(&client_rx, client_sock,
132                           sizeof(struct worker_reply));
133         if (!error) {
134             struct worker_reply *reply = client_rx.header.data;
135             reply->reply_cb(&client_rx.payload, client_rx.fds,
136                             client_rx.n_fds, reply->reply_aux);
137             rxbuf_clear(&client_rx);
138         } else if (error != EAGAIN) {
139             worker_broke();
140             VLOG_ABORT("receive from worker failed (%s)",
141                        ovs_retval_to_string(error));
142         }
143     }
144 }
145
146 /* Causes the poll loop to wake up if we need to process RPC replies. */
147 void
148 worker_wait(void)
149 {
150     if (worker_is_running()) {
151         poll_fd_wait(client_sock, POLLIN);
152     }
153 }
154 \f
155 /* Interface for main process to interact with the worker. */
156
157 /* Sends an RPC request to the worker process.  The worker process will call
158  * 'request_cb' passing the 'size' (zero or more) bytes of data in 'data' as
159  * arguments as well as the 'n_fds' (SOUTIL_MAX_FDS or fewer) file descriptors
160  * in 'fds'.
161  *
162  * If and only if 'reply_cb' is nonnull, 'request_cb' must call worker_reply()
163  * or worker_reply_iovec() with a reply.  The main process will later call
164  * 'reply_cb' with the reply data (if any) and file descriptors (if any).
165  *
166  * 'request_cb' receives copies (as if by dup()) of the file descriptors in
167  * fds[].  'request_cb' takes ownership of these copies, and the caller of
168  * worker_request() retains its ownership of the originals.
169  *
170  * This function may block until the RPC request has been sent (if the socket
171  * buffer fills up) but it does not wait for the reply (if any).  If this
172  * function blocks, it may invoke reply callbacks for previous requests.
173  *
174  * The worker process executes RPC requests in strict order of submission and
175  * runs each request to completion before beginning the next request.  The main
176  * process invokes reply callbacks in strict order of request submission. */
177 void
178 worker_request(const void *data, size_t size,
179                const int fds[], size_t n_fds,
180                worker_request_func *request_cb,
181                worker_reply_func *reply_cb, void *aux)
182 {
183     if (size > 0) {
184         struct iovec iov;
185
186         iov.iov_base = (void *) data;
187         iov.iov_len = size;
188         worker_request_iovec(&iov, 1, fds, n_fds, request_cb, reply_cb, aux);
189     } else {
190         worker_request_iovec(NULL, 0, fds, n_fds, request_cb, reply_cb, aux);
191     }
192 }
193
194 static int
195 worker_send_iovec(const struct iovec iovs[], size_t n_iovs,
196                   const int fds[], size_t n_fds)
197 {
198     size_t sent = 0;
199
200     for (;;) {
201         int error;
202
203         /* Try to send the rest of the request. */
204         error = send_iovec_and_fds_fully(client_sock, iovs, n_iovs,
205                                          fds, n_fds, sent, &sent);
206         if (error != EAGAIN) {
207             return error;
208         }
209
210         /* Process replies to avoid deadlock. */
211         worker_run();
212
213         poll_fd_wait(client_sock, POLLIN | POLLOUT);
214         poll_block();
215     }
216 }
217
218 /* Same as worker_request() except that the data to send is specified as an
219  * array of iovecs. */
220 void
221 worker_request_iovec(const struct iovec iovs[], size_t n_iovs,
222                      const int fds[], size_t n_fds,
223                      worker_request_func *request_cb,
224                      worker_reply_func *reply_cb, void *aux)
225 {
226     struct worker_request rq;
227     struct iovec *all_iovs;
228     int error;
229
230     assert(worker_is_running());
231
232     rq.request_len = iovec_len(iovs, n_iovs);
233     rq.request_cb = request_cb;
234     rq.reply_cb = reply_cb;
235     rq.reply_aux = aux;
236
237     all_iovs = prefix_iov(&rq, sizeof rq, iovs, n_iovs);
238     error = worker_send_iovec(all_iovs, n_iovs + 1, fds, n_fds);
239     if (error) {
240         worker_broke();
241         VLOG_ABORT("send failed (%s)", strerror(error));
242     }
243     free(all_iovs);
244 }
245
246 /* Closes the client socket, if any, so that worker_is_running() will return
247  * false.
248  *
249  * The client does this just before aborting if the worker process dies or
250  * malfunctions, to prevent the logging subsystem from trying to use the
251  * worker to log the failure. */
252 static void
253 worker_broke(void)
254 {
255     if (client_sock >= 0) {
256         close(client_sock);
257         client_sock = -1;
258     }
259 }
260 \f
261 /* Interfaces for RPC implementations (running in the worker process). */
262
263 static int server_sock = -1;
264 static bool expect_reply;
265 static struct worker_request request;
266
267 /* When a call to worker_request() or worker_request_iovec() provides a
268  * 'reply_cb' callback, the 'request_cb' implementation must call this function
269  * to send its reply.  The main process will call 'reply_cb' passing the
270  * 'size' (zero or more) bytes of data in 'data' as arguments as well as the
271  * 'n_fds' (SOUTIL_MAX_FDS or fewer) file descriptors in 'fds'.
272  *
273  * If a call to worker_request() or worker_request_iovec() provides no
274  * 'reply_cb' callback, the 'request_cb' implementation must not call this
275  * function.
276  *
277  * 'reply_cb' receives copies (as if by dup()) of the file descriptors in
278  * fds[].  'reply_cb' takes ownership of these copies, and the caller of
279  * worker_reply() retains its ownership of the originals.
280  *
281  * This function blocks until the RPC reply has been sent (if the socket buffer
282  * fills up) but it does not wait for the main process to receive or to process
283  * the reply. */
284 void
285 worker_reply(const void *data, size_t size, const int fds[], size_t n_fds)
286 {
287     if (size > 0) {
288         struct iovec iov;
289
290         iov.iov_base = (void *) data;
291         iov.iov_len = size;
292         worker_reply_iovec(&iov, 1, fds, n_fds);
293     } else {
294         worker_reply_iovec(NULL, 0, fds, n_fds);
295     }
296 }
297
298 /* Same as worker_reply() except that the data to send is specified as an array
299  * of iovecs. */
300 void
301 worker_reply_iovec(const struct iovec *iovs, size_t n_iovs,
302                        const int fds[], size_t n_fds)
303 {
304     struct worker_reply reply;
305     struct iovec *all_iovs;
306     int error;
307
308     assert(expect_reply);
309     expect_reply = false;
310
311     reply.reply_len = iovec_len(iovs, n_iovs);
312     reply.reply_cb = request.reply_cb;
313     reply.reply_aux = request.reply_aux;
314
315     all_iovs = prefix_iov(&reply, sizeof reply, iovs, n_iovs);
316
317     error = send_iovec_and_fds_fully_block(server_sock, all_iovs, n_iovs + 1,
318                                            fds, n_fds);
319     if (error == EPIPE) {
320         /* Parent probably died.  Continue processing any RPCs still buffered,
321          * to avoid missing log messages. */
322         VLOG_INFO("send failed (%s)", strerror(error));
323     } else if (error) {
324         VLOG_ABORT("send failed (%s)", strerror(error));
325     }
326
327     free(all_iovs);
328 }
329
330 static void
331 worker_main(int fd)
332 {
333     struct rxbuf rx;
334
335     server_sock = fd;
336
337     subprogram_name = "worker";
338     proctitle_set("%s: worker process for pid %lu",
339                   program_name, (unsigned long int) getppid());
340     VLOG_INFO("worker process started");
341
342     rxbuf_init(&rx);
343     for (;;) {
344         int error;
345
346         error = rxbuf_run(&rx, server_sock, sizeof(struct worker_request));
347         if (!error) {
348             request = *(struct worker_request *) rx.header.data;
349
350             expect_reply = request.reply_cb != NULL;
351             request.request_cb(&rx.payload, rx.fds, rx.n_fds);
352             assert(!expect_reply);
353
354             rxbuf_clear(&rx);
355         } else if (error == EOF && !rx.header.size) {
356             /* Main process closed the IPC socket.  Exit cleanly. */
357             break;
358         } else if (error != EAGAIN) {
359             VLOG_ABORT("RPC receive failed (%s)", strerror(error));
360         }
361
362         poll_fd_wait(server_sock, POLLIN);
363         poll_block();
364     }
365
366     VLOG_INFO("worker process exiting");
367     exit(0);
368 }
369 \f
370 static void
371 rxbuf_init(struct rxbuf *rx)
372 {
373     ofpbuf_init(&rx->header, 0);
374     rx->n_fds = 0;
375     ofpbuf_init(&rx->payload, 0);
376 }
377
378 static void
379 rxbuf_clear(struct rxbuf *rx)
380 {
381     ofpbuf_clear(&rx->header);
382     rx->n_fds = 0;
383     ofpbuf_clear(&rx->payload);
384 }
385
386 static int
387 rxbuf_run(struct rxbuf *rx, int sock, size_t header_len)
388 {
389     for (;;) {
390         if (!rx->header.size) {
391             int retval;
392
393             ofpbuf_clear(&rx->header);
394             ofpbuf_prealloc_tailroom(&rx->header, header_len);
395
396             retval = recv_data_and_fds(sock, rx->header.data, header_len,
397                                        rx->fds, &rx->n_fds);
398             if (retval <= 0) {
399                 return retval ? -retval : EOF;
400             }
401             rx->header.size += retval;
402         } else if (rx->header.size < header_len) {
403             size_t bytes_read;
404             int error;
405
406             error = read_fully(sock, ofpbuf_tail(&rx->header),
407                                header_len - rx->header.size, &bytes_read);
408             rx->header.size += bytes_read;
409             if (error) {
410                 return error;
411             }
412         } else {
413             size_t payload_len = *(size_t *) rx->header.data;
414
415             if (rx->payload.size < payload_len) {
416                 size_t left = payload_len - rx->payload.size;
417                 size_t bytes_read;
418                 int error;
419
420                 ofpbuf_prealloc_tailroom(&rx->payload, left);
421                 error = read_fully(sock, ofpbuf_tail(&rx->payload), left,
422                                    &bytes_read);
423                 rx->payload.size += bytes_read;
424                 if (error) {
425                     return error;
426                 }
427             } else {
428                 return 0;
429             }
430         }
431     }
432
433     return EAGAIN;
434 }
435
436 static struct iovec *
437 prefix_iov(void *data, size_t len, const struct iovec *iovs, size_t n_iovs)
438 {
439     struct iovec *dst;
440
441     dst = xmalloc((n_iovs + 1) * sizeof *dst);
442     dst[0].iov_base = data;
443     dst[0].iov_len = len;
444     memcpy(dst + 1, iovs, n_iovs * sizeof *iovs);
445
446     return dst;
447 }