5 #include "ntp_workimpl.h"
14 #include "ntp_stdlib.h"
15 #include "ntp_malloc.h"
16 #include "ntp_syslog.h"
19 #include "ntp_assert.h"
20 #include "ntp_unixtime.h"
21 #include "intreswork.h"
24 #define CHILD_MAX_IDLE (3 * 60) /* seconds, idle worker limit */
26 blocking_child ** blocking_children;
27 size_t blocking_children_alloc;
28 int worker_per_query; /* boolean */
29 int intres_req_pending;
32 #ifndef HAVE_IO_COMPLETION_PORT
36 * Provides an AF_UNIX socketpair on systems which have them, otherwise
37 * pair of unidirectional pipes.
49 #ifdef HAVE_SOCKETPAIR
50 rc = socketpair(AF_UNIX, SOCK_STREAM, 0, &fds[0]);
65 caller_fds[0] = fds[0];
66 caller_fds[1] = fds[1];
68 *is_pipe = called_pipe;
77 * Close all file descriptors except the given keep_fd.
86 for (fd = 0; fd < keep_fd; fd++)
89 close_all_beyond(keep_fd);
96 * Close all file descriptors after the given keep_fd, which is the
97 * highest fd to keep open.
104 # ifdef HAVE_CLOSEFROM
105 closefrom(keep_fd + 1);
106 # elif defined(F_CLOSEM)
108 * From 'Writing Reliable AIX Daemons,' SG24-4946-00,
109 * by Eric Agar (saves us from doing 32767 system
112 if (fcntl(keep_fd + 1, F_CLOSEM, 0) == -1)
113 msyslog(LOG_ERR, "F_CLOSEM(%d): %m", keep_fd + 1);
114 # else /* !HAVE_CLOSEFROM && !F_CLOSEM follows */
118 max_fd = GETDTABLESIZE();
119 for (fd = keep_fd + 1; fd < max_fd; fd++)
121 # endif /* !HAVE_CLOSEFROM && !F_CLOSEM */
123 #endif /* HAVE_IO_COMPLETION_PORT */
127 available_blocking_child_slot(void)
129 const size_t each = sizeof(blocking_children[0]);
136 for (slot = 0; slot < blocking_children_alloc; slot++) {
137 if (NULL == blocking_children[slot])
139 if (blocking_children[slot]->reusable) {
140 blocking_children[slot]->reusable = FALSE;
145 prev_alloc = blocking_children_alloc;
146 prev_octets = prev_alloc * each;
147 new_alloc = blocking_children_alloc + 4;
148 octets = new_alloc * each;
149 blocking_children = erealloc_zero(blocking_children, octets,
151 blocking_children_alloc = new_alloc;
158 queue_blocking_request(
159 blocking_work_req rtype,
162 blocking_work_callback done_func,
166 static u_int intres_slot = UINT_MAX;
169 blocking_pipe_header req_hdr;
171 req_hdr.octets = sizeof(req_hdr) + reqsize;
172 req_hdr.magic_sig = BLOCKING_REQ_MAGIC;
173 req_hdr.rtype = rtype;
174 req_hdr.done_func = done_func;
175 req_hdr.context = context;
177 child_slot = UINT_MAX;
178 if (worker_per_query || UINT_MAX == intres_slot ||
179 blocking_children[intres_slot]->reusable)
180 child_slot = available_blocking_child_slot();
181 if (!worker_per_query) {
182 if (UINT_MAX == intres_slot)
183 intres_slot = child_slot;
185 child_slot = intres_slot;
186 if (0 == intres_req_pending)
187 intres_timeout_req(0);
189 intres_req_pending++;
190 INSIST(UINT_MAX != child_slot);
191 c = blocking_children[child_slot];
193 c = emalloc_zero(sizeof(*c));
195 c->req_read_pipe = -1;
196 c->req_write_pipe = -1;
199 c->resp_read_pipe = -1;
200 c->resp_write_pipe = -1;
202 blocking_children[child_slot] = c;
204 req_hdr.child_idx = child_slot;
206 return send_blocking_req_internal(c, &req_hdr, req);
210 int queue_blocking_response(
212 blocking_pipe_header * resp,
214 const blocking_pipe_header * req
217 resp->octets = respsize;
218 resp->magic_sig = BLOCKING_RESP_MAGIC;
219 resp->rtype = req->rtype;
220 resp->context = req->context;
221 resp->done_func = req->done_func;
223 return send_blocking_resp_internal(c, resp);
228 process_blocking_resp(
232 blocking_pipe_header * resp;
236 * On Windows send_blocking_resp_internal() may signal the
237 * blocking_response_ready event multiple times while we're
238 * processing a response, so always consume all available
239 * responses before returning to test the event again.
244 resp = receive_blocking_resp_internal(c);
246 DEBUG_REQUIRE(BLOCKING_RESP_MAGIC ==
248 data = (char *)resp + sizeof(*resp);
249 intres_req_pending--;
250 (*resp->done_func)(resp->rtype, resp->context,
251 resp->octets - sizeof(*resp),
256 } while (NULL != resp);
258 if (!worker_per_query && 0 == intres_req_pending)
259 intres_timeout_req(CHILD_MAX_IDLE);
260 else if (worker_per_query)
266 * blocking_child_common runs as a forked child or a thread
269 blocking_child_common(
274 blocking_pipe_header *req;
278 req = receive_blocking_req_internal(c);
284 DEBUG_REQUIRE(BLOCKING_REQ_MAGIC == req->magic_sig);
286 switch (req->rtype) {
287 case BLOCKING_GETADDRINFO:
288 if (blocking_getaddrinfo(c, req))
292 case BLOCKING_GETNAMEINFO:
293 if (blocking_getnameinfo(c, req))
298 msyslog(LOG_ERR, "unknown req %d to blocking worker", req->rtype);
310 * worker_idle_timer_fired()
312 * The parent starts this timer when the last pending response has been
313 * received from the child, making it idle, and clears the timer when a
314 * request is dispatched to the child. Once the timer expires, the
315 * child is sent packing.
317 * This is called when worker_idle_timer is nonzero and less than or
318 * equal to current_time.
321 worker_idle_timer_fired(void)
326 DEBUG_REQUIRE(0 == intres_req_pending);
328 intres_timeout_req(0);
329 for (idx = 0; idx < blocking_children_alloc; idx++) {
330 c = blocking_children[idx];
338 #else /* !WORKER follows */
339 int ntp_worker_nonempty_compilation_unit;