1 /* Copyright (c) 2012, 2013 Nicira, Inc.
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:
7 * http://www.apache.org/licenses/LICENSE-2.0
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.
24 #include <sys/socket.h>
25 #include <sys/types.h>
30 #include "command-line.h"
33 #include "poll-loop.h"
34 #include "socket-util.h"
38 VLOG_DEFINE_THIS_MODULE(worker);
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'. */
48 /* Header for an RPC 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'. */
55 /* Receive buffer for a RPC request or reply. */
58 struct ofpbuf header; /* Header data. */
59 int fds[SOUTIL_MAX_FDS]; /* File descriptors. */
63 struct ofpbuf payload; /* Payload data. */
66 static int client_sock = -1;
67 static struct rxbuf client_rx;
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);
73 static struct iovec *prefix_iov(void *data, size_t len,
74 const struct iovec *iovs, size_t n_iovs);
76 static void worker_broke(void);
78 static void worker_main(int fd) NO_RETURN;
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
84 * The client should call worker_run() and worker_wait() from its main loop.
86 * Call this function between daemonize_start() and daemonize_complete(). */
92 assert(client_sock < 0);
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]);
99 if (!fork_and_clean_up()) {
100 /* In child (worker) process. */
101 daemonize_post_detach();
103 worker_main(work_fds[1]);
107 /* In parent (main) process. */
109 client_sock = work_fds[0];
110 rxbuf_init(&client_rx);
113 /* Returns true if this process has started a worker and the worker is not
114 * known to have malfunctioned. */
116 worker_is_running(void)
118 return client_sock >= 0;
121 /* If a worker process was started, processes RPC replies from it, calling the
122 * registered 'reply_cb' callbacks.
124 * If the worker process died or malfunctioned, aborts. */
128 if (worker_is_running()) {
131 error = rxbuf_run(&client_rx, client_sock,
132 sizeof(struct worker_reply));
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) {
140 VLOG_ABORT("receive from worker failed (%s)",
141 ovs_retval_to_string(error));
146 /* Causes the poll loop to wake up if we need to process RPC replies. */
150 if (worker_is_running()) {
151 poll_fd_wait(client_sock, POLLIN);
155 /* Interface for main process to interact with the worker. */
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
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).
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.
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.
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. */
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)
186 iov.iov_base = (void *) data;
188 worker_request_iovec(&iov, 1, fds, n_fds, request_cb, reply_cb, aux);
190 worker_request_iovec(NULL, 0, fds, n_fds, request_cb, reply_cb, aux);
195 worker_send_iovec(const struct iovec iovs[], size_t n_iovs,
196 const int fds[], size_t n_fds)
204 /* Try to send the rest of the request. */
205 error = send_iovec_and_fds_fully(client_sock, iovs, n_iovs,
206 fds, n_fds, sent, &sent);
207 if (error != EAGAIN) {
211 /* Process replies to avoid deadlock. */
214 /* Wait for 'client_sock' to become ready before trying again. We
215 * can't use poll_block() because it sometimes calls into vlog, which
216 * calls indirectly into worker_send_iovec(). To be usable here,
217 * poll_block() would therefore need to be reentrant, but it isn't
218 * (calling it recursively causes memory corruption and an eventual
220 pfd.fd = client_sock;
221 pfd.events = POLLIN | POLLOUT;
223 error = poll(&pfd, 1, -1) < 0 ? errno : 0;
224 } while (error == EINTR);
227 VLOG_ABORT("poll failed (%s)", strerror(error));
232 /* Same as worker_request() except that the data to send is specified as an
233 * array of iovecs. */
235 worker_request_iovec(const struct iovec iovs[], size_t n_iovs,
236 const int fds[], size_t n_fds,
237 worker_request_func *request_cb,
238 worker_reply_func *reply_cb, void *aux)
240 static bool recursing = false;
241 struct worker_request rq;
242 struct iovec *all_iovs;
245 assert(worker_is_running());
249 rq.request_len = iovec_len(iovs, n_iovs);
250 rq.request_cb = request_cb;
251 rq.reply_cb = reply_cb;
254 all_iovs = prefix_iov(&rq, sizeof rq, iovs, n_iovs);
255 error = worker_send_iovec(all_iovs, n_iovs + 1, fds, n_fds);
258 VLOG_ABORT("send failed (%s)", strerror(error));
265 /* Closes the client socket, if any, so that worker_is_running() will return
268 * The client does this just before aborting if the worker process dies or
269 * malfunctions, to prevent the logging subsystem from trying to use the
270 * worker to log the failure. */
274 if (client_sock >= 0) {
280 /* Interfaces for RPC implementations (running in the worker process). */
282 static int server_sock = -1;
283 static bool expect_reply;
284 static struct worker_request request;
286 /* When a call to worker_request() or worker_request_iovec() provides a
287 * 'reply_cb' callback, the 'request_cb' implementation must call this function
288 * to send its reply. The main process will call 'reply_cb' passing the
289 * 'size' (zero or more) bytes of data in 'data' as arguments as well as the
290 * 'n_fds' (SOUTIL_MAX_FDS or fewer) file descriptors in 'fds'.
292 * If a call to worker_request() or worker_request_iovec() provides no
293 * 'reply_cb' callback, the 'request_cb' implementation must not call this
296 * 'reply_cb' receives copies (as if by dup()) of the file descriptors in
297 * fds[]. 'reply_cb' takes ownership of these copies, and the caller of
298 * worker_reply() retains its ownership of the originals.
300 * This function blocks until the RPC reply has been sent (if the socket buffer
301 * fills up) but it does not wait for the main process to receive or to process
304 worker_reply(const void *data, size_t size, const int fds[], size_t n_fds)
309 iov.iov_base = (void *) data;
311 worker_reply_iovec(&iov, 1, fds, n_fds);
313 worker_reply_iovec(NULL, 0, fds, n_fds);
317 /* Same as worker_reply() except that the data to send is specified as an array
320 worker_reply_iovec(const struct iovec *iovs, size_t n_iovs,
321 const int fds[], size_t n_fds)
323 struct worker_reply reply;
324 struct iovec *all_iovs;
327 assert(expect_reply);
328 expect_reply = false;
330 reply.reply_len = iovec_len(iovs, n_iovs);
331 reply.reply_cb = request.reply_cb;
332 reply.reply_aux = request.reply_aux;
334 all_iovs = prefix_iov(&reply, sizeof reply, iovs, n_iovs);
336 error = send_iovec_and_fds_fully_block(server_sock, all_iovs, n_iovs + 1,
338 if (error == EPIPE) {
339 /* Parent probably died. Continue processing any RPCs still buffered,
340 * to avoid missing log messages. */
341 VLOG_INFO("send failed (%s)", strerror(error));
343 VLOG_ABORT("send failed (%s)", strerror(error));
356 subprogram_name = "worker";
357 proctitle_set("worker process for pid %lu", (unsigned long int) getppid());
358 VLOG_INFO("worker process started");
364 error = rxbuf_run(&rx, server_sock, sizeof(struct worker_request));
366 request = *(struct worker_request *) rx.header.data;
368 expect_reply = request.reply_cb != NULL;
369 request.request_cb(&rx.payload, rx.fds, rx.n_fds);
370 assert(!expect_reply);
373 } else if (error == EOF && !rx.header.size) {
374 /* Main process closed the IPC socket. Exit cleanly. */
376 } else if (error != EAGAIN) {
377 VLOG_ABORT("RPC receive failed (%s)", strerror(error));
380 poll_fd_wait(server_sock, POLLIN);
384 VLOG_INFO("worker process exiting");
389 rxbuf_init(struct rxbuf *rx)
391 ofpbuf_init(&rx->header, 0);
393 ofpbuf_init(&rx->payload, 0);
397 rxbuf_clear(struct rxbuf *rx)
399 ofpbuf_clear(&rx->header);
401 ofpbuf_clear(&rx->payload);
405 rxbuf_run(struct rxbuf *rx, int sock, size_t header_len)
408 if (!rx->header.size) {
411 ofpbuf_clear(&rx->header);
412 ofpbuf_prealloc_tailroom(&rx->header, header_len);
414 retval = recv_data_and_fds(sock, rx->header.data, header_len,
415 rx->fds, &rx->n_fds);
417 return retval ? -retval : EOF;
419 rx->header.size += retval;
420 } else if (rx->header.size < header_len) {
424 error = read_fully(sock, ofpbuf_tail(&rx->header),
425 header_len - rx->header.size, &bytes_read);
426 rx->header.size += bytes_read;
431 size_t payload_len = *(size_t *) rx->header.data;
433 if (rx->payload.size < payload_len) {
434 size_t left = payload_len - rx->payload.size;
438 ofpbuf_prealloc_tailroom(&rx->payload, left);
439 error = read_fully(sock, ofpbuf_tail(&rx->payload), left,
441 rx->payload.size += bytes_read;
452 static struct iovec *
453 prefix_iov(void *data, size_t len, const struct iovec *iovs, size_t n_iovs)
457 dst = xmalloc((n_iovs + 1) * sizeof *dst);
458 dst[0].iov_base = data;
459 dst[0].iov_len = len;
460 memcpy(dst + 1, iovs, n_iovs * sizeof *iovs);