Line | Count | Source |
1 | | /* |
2 | | * File descriptors management functions. |
3 | | * |
4 | | * Copyright 2000-2014 Willy Tarreau <w@1wt.eu> |
5 | | * |
6 | | * This program is free software; you can redistribute it and/or |
7 | | * modify it under the terms of the GNU General Public License |
8 | | * as published by the Free Software Foundation; either version |
9 | | * 2 of the License, or (at your option) any later version. |
10 | | * |
11 | | * There is no direct link between the FD and the updates list. There is only a |
12 | | * bit in the fdtab[] to indicate than a file descriptor is already present in |
13 | | * the updates list. Once an fd is present in the updates list, it will have to |
14 | | * be considered even if its changes are reverted in the middle or if the fd is |
15 | | * replaced. |
16 | | * |
17 | | * The event state for an FD, as found in fdtab[].state, is maintained for each |
18 | | * direction. The state field is built this way, with R bits in the low nibble |
19 | | * and W bits in the high nibble for ease of access and debugging : |
20 | | * |
21 | | * 7 6 5 4 3 2 1 0 |
22 | | * [ 0 | 0 | RW | AW | 0 | 0 | RR | AR ] |
23 | | * |
24 | | * A* = active *R = read |
25 | | * R* = ready *W = write |
26 | | * |
27 | | * An FD is marked "active" when there is a desire to use it. |
28 | | * An FD is marked "ready" when it has not faced a new EAGAIN since last wake-up |
29 | | * (it is a cache of the last EAGAIN regardless of polling changes). Each poller |
30 | | * has its own "polled" state for the same fd, as stored in the polled_mask. |
31 | | * |
32 | | * We have 4 possible states for each direction based on these 2 flags : |
33 | | * |
34 | | * +---+---+----------+---------------------------------------------+ |
35 | | * | R | A | State | Description | |
36 | | * +---+---+----------+---------------------------------------------+ |
37 | | * | 0 | 0 | DISABLED | No activity desired, not ready. | |
38 | | * | 0 | 1 | ACTIVE | Activity desired. | |
39 | | * | 1 | 0 | STOPPED | End of activity. | |
40 | | * | 1 | 1 | READY | Activity desired and reported. | |
41 | | * +---+---+----------+---------------------------------------------+ |
42 | | * |
43 | | * The transitions are pretty simple : |
44 | | * - fd_want_*() : set flag A |
45 | | * - fd_stop_*() : clear flag A |
46 | | * - fd_cant_*() : clear flag R (when facing EAGAIN) |
47 | | * - fd_may_*() : set flag R (upon return from poll()) |
48 | | * |
49 | | * Each poller then computes its own polled state : |
50 | | * if (A) { if (!R) P := 1 } else { P := 0 } |
51 | | * |
52 | | * The state transitions look like the diagram below. |
53 | | * |
54 | | * may +----------+ |
55 | | * ,----| DISABLED | (READY=0, ACTIVE=0) |
56 | | * | +----------+ |
57 | | * | want | ^ |
58 | | * | | | |
59 | | * | v | stop |
60 | | * | +----------+ |
61 | | * | | ACTIVE | (READY=0, ACTIVE=1) |
62 | | * | +----------+ |
63 | | * | | ^ |
64 | | * | may | | |
65 | | * | v | EAGAIN (can't) |
66 | | * | +--------+ |
67 | | * | | READY | (READY=1, ACTIVE=1) |
68 | | * | +--------+ |
69 | | * | stop | ^ |
70 | | * | | | |
71 | | * | v | want |
72 | | * | +---------+ |
73 | | * `--->| STOPPED | (READY=1, ACTIVE=0) |
74 | | * +---------+ |
75 | | */ |
76 | | |
77 | | #include <stdio.h> |
78 | | #include <string.h> |
79 | | #include <unistd.h> |
80 | | #include <fcntl.h> |
81 | | #include <sys/types.h> |
82 | | #include <sys/resource.h> |
83 | | #include <sys/uio.h> |
84 | | |
85 | | #if defined(USE_POLL) |
86 | | #include <poll.h> |
87 | | #endif |
88 | | #include <errno.h> |
89 | | |
90 | | #include <haproxy/api.h> |
91 | | #include <haproxy/activity.h> |
92 | | #include <haproxy/cfgparse.h> |
93 | | #include <haproxy/fd.h> |
94 | | #include <haproxy/global.h> |
95 | | #include <haproxy/log.h> |
96 | | #include <haproxy/port_range.h> |
97 | | #include <haproxy/ticks.h> |
98 | | #include <haproxy/tools.h> |
99 | | |
100 | | |
101 | | THREAD_LOCAL struct fdtab *fdtab = NULL; /* array of all the file descriptors */ |
102 | | static struct fdtab *_fdtab; |
103 | | THREAD_LOCAL struct polled_mask *polled_mask = NULL; /* Array for the polled_mask of each fd */ |
104 | | static struct polled_mask *_polled_mask; |
105 | | int totalconn; /* total # of terminated sessions */ |
106 | | int actconn; /* # of active sessions */ |
107 | | |
108 | | struct poller pollers[MAX_POLLERS] __read_mostly; |
109 | | struct poller cur_poller __read_mostly; |
110 | | int nbpollers = 0; |
111 | | |
112 | | volatile struct fdlist update_list[MAX_TGROUPS]; // Global update list |
113 | | |
114 | | THREAD_LOCAL int *fd_updt = NULL; // FD updates list |
115 | | THREAD_LOCAL int fd_nbupdt = 0; // number of updates in the list |
116 | | THREAD_LOCAL int fd_highest = -1; // highest FD known by the current thread |
117 | | THREAD_LOCAL int poller_rd_pipe = -1; // Pipe to wake the thread |
118 | | int poller_wr_pipe[MAX_THREADS] __read_mostly; // Pipe to wake the threads |
119 | | |
120 | | volatile int ha_used_fds = 0; // Number of FD we're currently using |
121 | | |
122 | | /* adds fd <fd> to fd list <list> if it was not yet in it */ |
123 | | void fd_add_to_fd_list(volatile struct fdlist *list, int fd) |
124 | 0 | { |
125 | 0 | int next; |
126 | 0 | int new; |
127 | 0 | int old; |
128 | 0 | int last; |
129 | |
|
130 | 0 | redo_next: |
131 | 0 | next = HA_ATOMIC_LOAD(&fdtab[fd].update.next); |
132 | | /* Check that we're not already in the cache, and if not, lock us. */ |
133 | 0 | if (next > -2) |
134 | 0 | goto done; |
135 | 0 | if (next == -2) |
136 | 0 | goto redo_next; |
137 | 0 | if (!_HA_ATOMIC_CAS(&fdtab[fd].update.next, &next, -2)) |
138 | 0 | goto redo_next; |
139 | 0 | __ha_barrier_atomic_store(); |
140 | |
|
141 | 0 | new = fd; |
142 | 0 | redo_last: |
143 | | /* First, insert in the linked list */ |
144 | 0 | last = list->last; |
145 | 0 | old = -1; |
146 | |
|
147 | 0 | fdtab[fd].update.prev = -2; |
148 | | /* Make sure the "prev" store is visible before we update the last entry */ |
149 | 0 | __ha_barrier_store(); |
150 | |
|
151 | 0 | if (unlikely(last == -1)) { |
152 | | /* list is empty, try to add ourselves alone so that list->last=fd */ |
153 | 0 | if (unlikely(!_HA_ATOMIC_CAS(&list->last, &old, new))) |
154 | 0 | goto redo_last; |
155 | | |
156 | | /* list->first was necessary -1, we're guaranteed to be alone here */ |
157 | 0 | list->first = fd; |
158 | 0 | } else { |
159 | | /* adding ourselves past the last element |
160 | | * The CAS will only succeed if its next is -1, |
161 | | * which means it's in the cache, and the last element. |
162 | | */ |
163 | 0 | if (unlikely(!_HA_ATOMIC_CAS(&fdtab[last].update.next, &old, new))) |
164 | 0 | goto redo_last; |
165 | | |
166 | | /* Then, update the last entry */ |
167 | 0 | list->last = fd; |
168 | 0 | } |
169 | 0 | __ha_barrier_store(); |
170 | | /* since we're alone at the end of the list and still locked(-2), |
171 | | * we know no one tried to add past us. Mark the end of list. |
172 | | */ |
173 | 0 | fdtab[fd].update.prev = last; |
174 | 0 | fdtab[fd].update.next = -1; |
175 | 0 | __ha_barrier_store(); |
176 | 0 | done: |
177 | 0 | return; |
178 | 0 | } |
179 | | |
180 | | /* removes fd <fd> from fd list <list> */ |
181 | | void fd_rm_from_fd_list(volatile struct fdlist *list, int fd) |
182 | 0 | { |
183 | 0 | #if defined(HA_HAVE_CAS_DW) || defined(HA_CAS_IS_8B) |
184 | 0 | volatile union { |
185 | 0 | struct fdlist_entry ent; |
186 | 0 | uint64_t u64; |
187 | 0 | uint32_t u32[2]; |
188 | 0 | } cur_list, next_list; |
189 | 0 | #endif |
190 | 0 | int old; |
191 | 0 | int new = -2; |
192 | 0 | int prev; |
193 | 0 | int next; |
194 | 0 | int last; |
195 | 0 | lock_self: |
196 | 0 | #if (defined(HA_CAS_IS_8B) || defined(HA_HAVE_CAS_DW)) |
197 | 0 | next_list.ent.next = next_list.ent.prev = -2; |
198 | 0 | cur_list.ent = *(volatile typeof(fdtab->update)*)&fdtab[fd].update; |
199 | | /* First, attempt to lock our own entries */ |
200 | 0 | do { |
201 | | /* The FD is not in the FD cache, give up */ |
202 | 0 | if (unlikely(cur_list.ent.next <= -3)) |
203 | 0 | return; |
204 | 0 | if (unlikely(cur_list.ent.prev == -2 || cur_list.ent.next == -2)) |
205 | 0 | goto lock_self; |
206 | 0 | } while ( |
207 | 0 | #ifdef HA_CAS_IS_8B |
208 | 0 | unlikely(!_HA_ATOMIC_CAS(((uint64_t *)&fdtab[fd].update), (uint64_t *)&cur_list.u64, next_list.u64)) |
209 | | #else |
210 | | unlikely(!_HA_ATOMIC_DWCAS(((long *)&fdtab[fd].update), (uint32_t *)&cur_list.u32, (const uint32_t *)&next_list.u32)) |
211 | | #endif |
212 | 0 | ); |
213 | 0 | next = cur_list.ent.next; |
214 | 0 | prev = cur_list.ent.prev; |
215 | |
|
216 | | #else |
217 | | lock_self_next: |
218 | | next = HA_ATOMIC_LOAD(&fdtab[fd].update.next); |
219 | | if (next == -2) |
220 | | goto lock_self_next; |
221 | | if (next <= -3) |
222 | | goto done; |
223 | | if (unlikely(!_HA_ATOMIC_CAS(&fdtab[fd].update.next, &next, -2))) |
224 | | goto lock_self_next; |
225 | | lock_self_prev: |
226 | | prev = HA_ATOMIC_LOAD(&fdtab[fd].update.prev); |
227 | | if (prev == -2) |
228 | | goto lock_self_prev; |
229 | | if (unlikely(!_HA_ATOMIC_CAS(&fdtab[fd].update.prev, &prev, -2))) |
230 | | goto lock_self_prev; |
231 | | #endif |
232 | 0 | __ha_barrier_atomic_store(); |
233 | | |
234 | | /* Now, lock the entries of our neighbours */ |
235 | 0 | if (likely(prev != -1)) { |
236 | 0 | redo_prev: |
237 | 0 | old = fd; |
238 | |
|
239 | 0 | if (unlikely(!_HA_ATOMIC_CAS(&fdtab[prev].update.next, &old, new))) { |
240 | 0 | if (unlikely(old == -2)) { |
241 | | /* Neighbour already locked, give up and |
242 | | * retry again once he's done |
243 | | */ |
244 | 0 | fdtab[fd].update.prev = prev; |
245 | 0 | __ha_barrier_store(); |
246 | 0 | fdtab[fd].update.next = next; |
247 | 0 | __ha_barrier_store(); |
248 | 0 | goto lock_self; |
249 | 0 | } |
250 | 0 | goto redo_prev; |
251 | 0 | } |
252 | 0 | } |
253 | 0 | if (likely(next != -1)) { |
254 | 0 | redo_next: |
255 | 0 | old = fd; |
256 | 0 | if (unlikely(!_HA_ATOMIC_CAS(&fdtab[next].update.prev, &old, new))) { |
257 | 0 | if (unlikely(old == -2)) { |
258 | | /* Neighbour already locked, give up and |
259 | | * retry again once he's done |
260 | | */ |
261 | 0 | if (prev != -1) { |
262 | 0 | fdtab[prev].update.next = fd; |
263 | 0 | __ha_barrier_store(); |
264 | 0 | } |
265 | 0 | fdtab[fd].update.prev = prev; |
266 | 0 | __ha_barrier_store(); |
267 | 0 | fdtab[fd].update.next = next; |
268 | 0 | __ha_barrier_store(); |
269 | 0 | goto lock_self; |
270 | 0 | } |
271 | 0 | goto redo_next; |
272 | 0 | } |
273 | 0 | } |
274 | 0 | if (list->first == fd) |
275 | 0 | list->first = next; |
276 | 0 | __ha_barrier_store(); |
277 | 0 | last = list->last; |
278 | 0 | while (unlikely(last == fd && (!_HA_ATOMIC_CAS(&list->last, &last, prev)))) |
279 | 0 | __ha_compiler_barrier(); |
280 | | /* Make sure we let other threads know we're no longer in cache, |
281 | | * before releasing our neighbours. |
282 | | */ |
283 | 0 | __ha_barrier_store(); |
284 | 0 | if (likely(prev != -1)) |
285 | 0 | fdtab[prev].update.next = next; |
286 | 0 | __ha_barrier_store(); |
287 | 0 | if (likely(next != -1)) |
288 | 0 | fdtab[next].update.prev = prev; |
289 | 0 | __ha_barrier_store(); |
290 | | /* Ok, now we're out of the fd cache */ |
291 | 0 | fdtab[fd].update.next = -(next + 4); |
292 | 0 | __ha_barrier_store(); |
293 | 0 | done: |
294 | 0 | return; |
295 | 0 | } |
296 | | |
297 | | /* deletes the FD once nobody uses it anymore, as detected by the caller by its |
298 | | * thread_mask being zero and its running mask turning to zero. There is no |
299 | | * protection against concurrent accesses, it's up to the caller to make sure |
300 | | * only the last thread will call it. If called under isolation, it is safe to |
301 | | * call this from another group than the FD's. This is only for internal use, |
302 | | * please use fd_delete() instead. |
303 | | */ |
304 | | void _fd_delete_orphan(int fd) |
305 | 0 | { |
306 | 0 | int tgrp = fd_tgid(fd); |
307 | 0 | uint fd_disown; |
308 | |
|
309 | 0 | fd_disown = fdtab[fd].state & FD_DISOWN; |
310 | 0 | if (fdtab[fd].state & FD_LINGER_RISK) { |
311 | | /* this is generally set when connecting to servers */ |
312 | 0 | DISGUISE(setsockopt(fd, SOL_SOCKET, SO_LINGER, |
313 | 0 | (struct linger *) &nolinger, sizeof(struct linger))); |
314 | 0 | } |
315 | | |
316 | | /* It's expected that a close() will result in the FD disappearing from |
317 | | * pollers, but some pollers may have some internal bookkeeping to be |
318 | | * done prior to the call (e.g. remove references from internal tables). |
319 | | */ |
320 | 0 | if (cur_poller.clo) |
321 | 0 | cur_poller.clo(fd); |
322 | | |
323 | | /* now we're about to reset some of this FD's fields. We don't want |
324 | | * anyone to grab it anymore and we need to make sure those which could |
325 | | * possibly have stumbled upon it right now are leaving before we |
326 | | * proceed. This is done in two steps. First we reset the tgid so that |
327 | | * fd_take_tgid() and fd_grab_tgid() fail, then we wait for existing |
328 | | * ref counts to drop. Past this point we're alone dealing with the |
329 | | * FD's thead/running/update/polled masks. |
330 | | */ |
331 | 0 | fd_reset_tgid(fd); |
332 | |
|
333 | 0 | while (_HA_ATOMIC_LOAD(&fdtab[fd].refc_tgid) != 0) // refc==0 ? |
334 | 0 | __ha_cpu_relax(); |
335 | | |
336 | | /* we don't want this FD anymore in the global list */ |
337 | 0 | fd_rm_from_fd_list(&update_list[tgrp - 1], fd); |
338 | | |
339 | | /* no more updates on this FD are relevant anymore */ |
340 | 0 | HA_ATOMIC_STORE(&fdtab[fd].update_mask, 0); |
341 | 0 | if (fd_nbupdt > 0 && fd_updt[fd_nbupdt - 1] == fd) |
342 | 0 | fd_nbupdt--; |
343 | |
|
344 | 0 | if (unlikely(fdtab[fd].state & FD_HAS_PORT)) { |
345 | 0 | struct sockaddr_storage sa; |
346 | 0 | socklen_t addrlen = sizeof(sa); |
347 | 0 | int port; |
348 | |
|
349 | 0 | BUG_ON(!(fdtab[fd].state & FD_OWNER_PR)); |
350 | | /* |
351 | | * We have to release the port, let's use getsockname() |
352 | | * to figure out what the port was. |
353 | | */ |
354 | 0 | BUG_ON(getsockname(fd, (struct sockaddr *)&sa, &addrlen) != 0); |
355 | 0 | port = get_host_port(&sa); |
356 | 0 | if (!port) |
357 | 0 | ABORT_NOW(); |
358 | 0 | port_range_release_port(fdtab[fd].owner, port); |
359 | 0 | } |
360 | | |
361 | 0 | polled_mask[fd].poll_recv = polled_mask[fd].poll_send = 0; |
362 | |
|
363 | 0 | fdtab[fd].state = 0; |
364 | |
|
365 | | #ifdef DEBUG_FD |
366 | | fdtab[fd].event_count = 0; |
367 | | #endif |
368 | 0 | fdtab[fd].owner = NULL; |
369 | | |
370 | | /* perform the close() call last as it's what unlocks the instant reuse |
371 | | * of this FD by any other thread. |
372 | | */ |
373 | 0 | if (!fd_disown) { |
374 | 0 | fdtab[fd].generation++; |
375 | 0 | close(fd); |
376 | 0 | } |
377 | 0 | _HA_ATOMIC_DEC(&ha_used_fds); |
378 | 0 | } |
379 | | |
380 | | /* Deletes an FD from the fdsets. The file descriptor is also closed, possibly |
381 | | * asynchronously. It is safe to call it from another thread from the same |
382 | | * group as the FD's or from a thread from a different group. |
383 | | * Calling from a thread from another group is handled by grabbing the |
384 | | * reference count, adding the FD_MUST_CLOSE and doing the close ourselves |
385 | | * only if there is no other thread currently using the fd. |
386 | | */ |
387 | | void fd_delete(int fd) |
388 | 0 | { |
389 | | /* This must never happen and would definitely indicate a bug, in |
390 | | * addition to overwriting some unexpected memory areas. |
391 | | */ |
392 | 0 | BUG_ON(fd < 0 || fd >= global.maxsock); |
393 | | |
394 | | /* NOTE: The master when going into reexec mode re-closes all FDs after |
395 | | * they were already dispatched. But we know we didn't start the polling |
396 | | * threads so we can still close them. The masks will probably not match |
397 | | * however so we force the value and erase the refcount if any. |
398 | | */ |
399 | 0 | if (unlikely(global.mode & MODE_STARTING)) |
400 | 0 | fdtab[fd].refc_tgid = ti->tgid; |
401 | | |
402 | | /* the tgid cannot change before a complete close so we should never |
403 | | * face the situation where we try to close an fd that was reassigned. |
404 | | * However there is one corner case where this happens, it's when an |
405 | | * attempt to pause a listener fails (e.g. abns), leaving the listener |
406 | | * in fault state and it is forcefully stopped. |
407 | | * To make it work when the fd does not belong to our thread group, |
408 | | * first increase the tgid refcount, so that we're sure nobody will |
409 | | * close it before we figure out if we should, then set the |
410 | | * FD_MUST_CLOSE bit, so that any running thread will take care |
411 | | * of closing it. |
412 | | * Once we set the thread_mask to 0, we know for sure that if |
413 | | * nobody is running, nobody ever will, so at this point, if |
414 | | * there is no running thread, and the FD_MUST_CLOSE bit is still |
415 | | * set, we know it is safe to actually close the fd. |
416 | | */ |
417 | 0 | if (unlikely(fd_tgid(fd) != ti->tgid)) { |
418 | 0 | uint tgrp; |
419 | |
|
420 | 0 | tgrp = fd_take_tgid(fd); |
421 | |
|
422 | 0 | HA_ATOMIC_OR(&fdtab[fd].state, FD_MUST_CLOSE); |
423 | 0 | HA_ATOMIC_STORE(&fdtab[fd].thread_mask, 0); |
424 | |
|
425 | 0 | if (!HA_ATOMIC_LOAD(&fdtab[fd].running_mask) && |
426 | 0 | HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) { |
427 | 0 | if (tgrp) |
428 | 0 | fd_drop_tgid(fd); |
429 | 0 | _fd_delete_orphan(fd); |
430 | 0 | } |
431 | 0 | else if (tgrp) |
432 | 0 | fd_drop_tgid(fd); |
433 | 0 | return; |
434 | 0 | } |
435 | | |
436 | | /* we must postpone removal of an FD that may currently be in use |
437 | | * by another thread. This can happen in the following two situations: |
438 | | * - after a takeover, the owning thread closes the connection but |
439 | | * the previous one just woke up from the poller and entered |
440 | | * the FD handler iocb. That thread holds an entry in running_mask |
441 | | * and requires removal protection. |
442 | | * - multiple threads are accepting connections on a listener, and |
443 | | * one of them (or even an separate one) decides to unbind the |
444 | | * listener under the listener's lock while other ones still hold |
445 | | * the running bit. |
446 | | * In both situations the FD is marked as unused (thread_mask = 0) and |
447 | | * will not take new bits in its running_mask so we have the guarantee |
448 | | * that the last thread eliminating running_mask is the one allowed to |
449 | | * safely delete the FD. Most of the time it will be the current thread. |
450 | | * We still need to set and check the one-shot flag FD_MUST_CLOSE |
451 | | * to take care of the rare cases where a thread wakes up on late I/O |
452 | | * before the thread_mask is zero, and sets its bit in the running_mask |
453 | | * just after the current thread finishes clearing its own bit, hence |
454 | | * the two threads see themselves as last ones (which they really are). |
455 | | */ |
456 | | |
457 | 0 | HA_ATOMIC_OR(&fdtab[fd].running_mask, ti->ltid_bit); |
458 | 0 | HA_ATOMIC_OR(&fdtab[fd].state, FD_MUST_CLOSE); |
459 | 0 | HA_ATOMIC_STORE(&fdtab[fd].thread_mask, 0); |
460 | 0 | if (fd_clr_running(fd) == ti->ltid_bit) { |
461 | 0 | if (HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) { |
462 | 0 | _fd_delete_orphan(fd); |
463 | 0 | } |
464 | 0 | } |
465 | 0 | } |
466 | | |
467 | | /* makes the new fd non-blocking and clears all other O_* flags; this is meant |
468 | | * to be used on new FDs. Returns -1 on failure. The result is disguised at the |
469 | | * end because some callers need to be able to ignore it regardless of the libc |
470 | | * attributes. |
471 | | */ |
472 | | int fd_set_nonblock(int fd) |
473 | 0 | { |
474 | 0 | int ret = fcntl(fd, F_SETFL, O_NONBLOCK); |
475 | |
|
476 | 0 | return DISGUISE(ret); |
477 | 0 | } |
478 | | |
479 | | /* sets the close-on-exec flag on fd; returns -1 on failure. The result is |
480 | | * disguised at the end because some callers need to be able to ignore it |
481 | | * regardless of the libc attributes. |
482 | | */ |
483 | | int fd_set_cloexec(int fd) |
484 | 0 | { |
485 | 0 | int flags, ret; |
486 | |
|
487 | 0 | flags = fcntl(fd, F_GETFD); |
488 | 0 | flags |= FD_CLOEXEC; |
489 | 0 | ret = fcntl(fd, F_SETFD, flags); |
490 | 0 | return DISGUISE(ret); |
491 | 0 | } |
492 | | |
493 | | /* Migrate a FD to a new thread <new_tid>. It is explicitly permitted to |
494 | | * migrate to another thread group, the function takes the necessary locking |
495 | | * for this. It is even permitted to migrate from a foreign group to another, |
496 | | * but the calling thread must be certain that the FD is not about to close |
497 | | * when doing so, reason why it is highly recommended that only one of the |
498 | | * FD's owners performs this operation. The polling is completely disabled. |
499 | | * The operation never fails. |
500 | | */ |
501 | | void fd_migrate_on(int fd, uint new_tid) |
502 | 0 | { |
503 | 0 | struct thread_info *new_ti = &ha_thread_info[new_tid]; |
504 | | |
505 | | /* we must be alone to work on this idle FD. If not, it means that its |
506 | | * poller is currently waking up and is about to use it, likely to |
507 | | * close it on shut/error, but maybe also to process any unexpectedly |
508 | | * pending data. It's also possible that the FD was closed and |
509 | | * reassigned to another thread group, so let's be careful. |
510 | | */ |
511 | 0 | fd_lock_tgid(fd, new_ti->tgid); |
512 | | |
513 | | /* now we have exclusive access to it. From now FD belongs to tid_bit |
514 | | * for this tgid. |
515 | | */ |
516 | 0 | HA_ATOMIC_STORE(&fdtab[fd].thread_mask, new_ti->ltid_bit); |
517 | | |
518 | | /* Make sure the FD doesn't have the active bit. It is possible that |
519 | | * the fd is polled by the thread that used to own it, the new thread |
520 | | * is supposed to call subscribe() later, to activate polling. |
521 | | */ |
522 | 0 | fd_stop_both(fd); |
523 | | |
524 | | /* we're done with it. As soon as we unlock it, other threads from the |
525 | | * target group can manipulate it. However it may only disappear once |
526 | | * we drop the reference. |
527 | | */ |
528 | 0 | fd_unlock_tgid(fd); |
529 | 0 | fd_drop_tgid(fd); |
530 | 0 | } |
531 | | |
532 | | /* |
533 | | * Take over a FD belonging to another thread. |
534 | | * unexpected_conn is the expected owner of the fd. |
535 | | * Returns 0 on success, and -1 on failure. |
536 | | */ |
537 | | int fd_takeover(int fd, void *expected_owner) |
538 | 0 | { |
539 | 0 | unsigned long old; |
540 | 0 | int changing_tgid = 0; |
541 | 0 | int old_ltid, old_tgid; |
542 | | |
543 | | /* protect ourself against a delete then an insert for the same fd, |
544 | | * if it happens, then the owner will no longer be the expected |
545 | | * connection. |
546 | | */ |
547 | 0 | if (fdtab[fd].owner != expected_owner) |
548 | 0 | return -1; |
549 | | |
550 | | /* We're taking a connection from a different thread group */ |
551 | 0 | if ((fdtab[fd].refc_tgid & 0x7fff) != tgid) { |
552 | 0 | BUG_ON(global.tune.options & GTUNE_NO_TG_FD_SHARING); |
553 | |
|
554 | 0 | changing_tgid = 1; |
555 | |
|
556 | 0 | old_tgid = fd_tgid(fd); |
557 | 0 | BUG_ON(atleast2(fdtab[fd].thread_mask)); |
558 | 0 | old_ltid = my_ffsl(fdtab[fd].thread_mask) - 1; |
559 | |
|
560 | 0 | if (unlikely(!fd_lock_tgid_cur(fd))) |
561 | 0 | return -1; |
562 | 0 | } else { |
563 | | /* we must be alone to work on this idle FD. If not, it means that its |
564 | | * poller is currently waking up and is about to use it, likely to |
565 | | * close it on shut/error, but maybe also to process any unexpectedly |
566 | | * pending data. It's also possible that the FD was closed and |
567 | | * reassigned to another thread group, so let's be careful. |
568 | | */ |
569 | 0 | if (unlikely(!fd_grab_tgid(fd, ti->tgid))) |
570 | 0 | return -1; |
571 | 0 | } |
572 | | |
573 | 0 | old = 0; |
574 | 0 | if (!HA_ATOMIC_CAS(&fdtab[fd].running_mask, &old, ti->ltid_bit)) { |
575 | 0 | if (changing_tgid) |
576 | 0 | fd_unlock_tgid(fd); |
577 | 0 | fd_drop_tgid(fd); |
578 | 0 | return -1; |
579 | 0 | } |
580 | | |
581 | | /* success, from now on it's ours */ |
582 | 0 | HA_ATOMIC_STORE(&fdtab[fd].thread_mask, ti->ltid_bit); |
583 | | |
584 | | /* |
585 | | * Change the tgid to our own tgid. |
586 | | * This removes the lock, we don't need it anymore, but we keep |
587 | | * the refcount. |
588 | | */ |
589 | 0 | if (changing_tgid) { |
590 | 0 | fd_update_tgid(fd, tgid); |
591 | 0 | if (cur_poller.fixup_tgid_takeover) |
592 | 0 | cur_poller.fixup_tgid_takeover(&cur_poller, fd, old_ltid, old_tgid); |
593 | 0 | } |
594 | | |
595 | | /* Make sure the FD doesn't have the active bit. It is possible that |
596 | | * the fd is polled by the thread that used to own it, the new thread |
597 | | * is supposed to call subscribe() later, to activate polling. |
598 | | */ |
599 | 0 | fd_stop_recv(fd); |
600 | | |
601 | | /* essentially for debugging */ |
602 | 0 | fdtab[fd].nb_takeover++; |
603 | | |
604 | | /* we're done with it */ |
605 | 0 | HA_ATOMIC_AND(&fdtab[fd].running_mask, ~ti->ltid_bit); |
606 | | |
607 | | /* no more changes planned */ |
608 | 0 | fd_drop_tgid(fd); |
609 | 0 | return 0; |
610 | 0 | } |
611 | | |
612 | | void updt_fd_polling(const int fd) |
613 | 0 | { |
614 | 0 | uint tgrp = fd_take_tgid(fd); |
615 | | |
616 | | /* closed ? may happen */ |
617 | 0 | if (!tgrp) |
618 | 0 | return; |
619 | | |
620 | 0 | if (unlikely(tgrp != tgid && tgrp <= MAX_TGROUPS)) { |
621 | | /* Hmmm delivered an update for another group... That may |
622 | | * happen on suspend/resume of a listener for example when |
623 | | * the FD was not even marked for running. Let's broadcast |
624 | | * the update. |
625 | | */ |
626 | 0 | unsigned long update_mask = fdtab[fd].update_mask; |
627 | 0 | int thr; |
628 | |
|
629 | 0 | while (!_HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask, |
630 | 0 | _HA_ATOMIC_LOAD(&ha_tgroup_info[tgrp - 1].threads_enabled))) |
631 | 0 | __ha_cpu_relax(); |
632 | |
|
633 | 0 | fd_add_to_fd_list(&update_list[tgrp - 1], fd); |
634 | |
|
635 | 0 | thr = one_among_mask(fdtab[fd].thread_mask & ha_tgroup_info[tgrp - 1].threads_enabled, |
636 | 0 | statistical_prng_range(ha_tgroup_info[tgrp - 1].count)); |
637 | 0 | thr += ha_tgroup_info[tgrp - 1].base; |
638 | 0 | wake_thread(thr); |
639 | |
|
640 | 0 | fd_drop_tgid(fd); |
641 | 0 | return; |
642 | 0 | } |
643 | | |
644 | 0 | fd_drop_tgid(fd); |
645 | |
|
646 | 0 | if (tg->threads_enabled == 1UL || (fdtab[fd].thread_mask & tg->threads_enabled) == ti->ltid_bit) { |
647 | 0 | if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid)) |
648 | 0 | return; |
649 | | |
650 | 0 | fd_updt[fd_nbupdt++] = fd; |
651 | 0 | } else { |
652 | 0 | unsigned long update_mask = fdtab[fd].update_mask; |
653 | 0 | do { |
654 | 0 | if (update_mask == fdtab[fd].thread_mask) // FIXME: this works only on thread-groups 1 |
655 | 0 | return; |
656 | 0 | } while (!_HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask, fdtab[fd].thread_mask)); |
657 | | |
658 | 0 | fd_add_to_fd_list(&update_list[tgid - 1], fd); |
659 | |
|
660 | 0 | if (fd_active(fd) && !(fdtab[fd].thread_mask & ti->ltid_bit)) { |
661 | | /* we need to wake up another thread to handle it immediately, any will fit, |
662 | | * so let's pick a random one so that it doesn't always end up on the same. |
663 | | */ |
664 | 0 | int thr = one_among_mask(fdtab[fd].thread_mask & tg->threads_enabled, |
665 | 0 | statistical_prng_range(tg->count)); |
666 | 0 | thr += tg->base; |
667 | 0 | wake_thread(thr); |
668 | 0 | } |
669 | 0 | } |
670 | 0 | } |
671 | | |
672 | | /* Update events seen for FD <fd> and its state if needed. This should be |
673 | | * called by the poller, passing FD_EV_*_{R,W,RW} in <evts>. FD_EV_ERR_* |
674 | | * doesn't need to also pass FD_EV_SHUT_*, it's implied. ERR and SHUT are |
675 | | * allowed to be reported regardless of R/W readiness. Returns one of |
676 | | * FD_UPDT_*. |
677 | | */ |
678 | | int fd_update_events(int fd, uint evts) |
679 | 0 | { |
680 | 0 | unsigned long locked; |
681 | 0 | uint old, new; |
682 | 0 | uint new_flags, must_stop; |
683 | 0 | ulong rmask, tmask; |
684 | |
|
685 | 0 | _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_STUCK); // this thread is still running |
686 | |
|
687 | 0 | if (unlikely(!fd_grab_tgid(fd, ti->tgid))) { |
688 | | /* the FD changed to another tgid, we can't safely |
689 | | * check it anymore. The bits in the masks are not |
690 | | * ours anymore and we're not allowed to touch them. |
691 | | * Ours have already been cleared and the FD was |
692 | | * closed in between so we can safely leave now. |
693 | | */ |
694 | 0 | activity[tid].poll_drop_fd++; |
695 | 0 | return FD_UPDT_CLOSED; |
696 | 0 | } |
697 | | |
698 | | /* Do not take running_mask if not strictly needed (will trigger a |
699 | | * cosmetic BUG_ON() in fd_insert() anyway if done). |
700 | | */ |
701 | 0 | tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask); |
702 | 0 | if (!(tmask & ti->ltid_bit)) |
703 | 0 | goto do_update; |
704 | | |
705 | 0 | HA_ATOMIC_OR(&fdtab[fd].running_mask, ti->ltid_bit); |
706 | | |
707 | | /* From this point, our bit may possibly be in thread_mask, but it may |
708 | | * still vanish, either because a takeover completed just before taking |
709 | | * the bit above with the new owner deleting the FD, or because a |
710 | | * takeover started just before taking the bit. In order to make sure a |
711 | | * started takeover is complete, we need to verify that all bits of |
712 | | * running_mask are present in thread_mask, since takeover first takes |
713 | | * running then atomically replaces thread_mask. Once it's stable, if |
714 | | * our bit remains there, no further takeover may happen because we |
715 | | * hold running, but if our bit is not there it means we've lost the |
716 | | * takeover race and have to decline touching the FD. Regarding the |
717 | | * risk of deletion, our bit in running_mask prevents fd_delete() from |
718 | | * finalizing the close, and the caller will leave the FD with a zero |
719 | | * thread_mask and the FD_MUST_CLOSE flag set. It will then be our |
720 | | * responsibility to close it. |
721 | | */ |
722 | 0 | do { |
723 | 0 | rmask = _HA_ATOMIC_LOAD(&fdtab[fd].running_mask); |
724 | 0 | tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask); |
725 | 0 | rmask &= ~ti->ltid_bit; |
726 | 0 | } while ((rmask & ~tmask) && (tmask & ti->ltid_bit)); |
727 | | |
728 | | /* Now tmask is stable. Do nothing if the FD was taken over under us */ |
729 | |
|
730 | 0 | if (!(tmask & ti->ltid_bit)) { |
731 | | /* a takeover has started */ |
732 | 0 | activity[tid].poll_skip_fd++; |
733 | |
|
734 | 0 | if (fd_clr_running(fd) == ti->ltid_bit) |
735 | 0 | goto closed_or_migrated; |
736 | | |
737 | 0 | goto do_update; |
738 | 0 | } |
739 | | |
740 | | /* with running we're safe now, we can drop the reference */ |
741 | 0 | fd_drop_tgid(fd); |
742 | |
|
743 | 0 | locked = (tmask != ti->ltid_bit); |
744 | | |
745 | | /* OK now we are guaranteed that our thread_mask was present and |
746 | | * that we're allowed to update the FD. |
747 | | */ |
748 | |
|
749 | 0 | new_flags = |
750 | 0 | ((evts & FD_EV_READY_R) ? FD_POLL_IN : 0) | |
751 | 0 | ((evts & FD_EV_READY_W) ? FD_POLL_OUT : 0) | |
752 | 0 | ((evts & FD_EV_SHUT_R) ? FD_POLL_HUP : 0) | |
753 | 0 | ((evts & FD_EV_ERR_RW) ? FD_POLL_ERR : 0); |
754 | | |
755 | | /* SHUTW reported while FD was active for writes is an error */ |
756 | 0 | if ((fdtab[fd].state & FD_EV_ACTIVE_W) && (evts & FD_EV_SHUT_W)) |
757 | 0 | new_flags |= FD_POLL_ERR; |
758 | | |
759 | | /* compute the inactive events reported late that must be stopped */ |
760 | 0 | must_stop = 0; |
761 | 0 | if (unlikely(!fd_active(fd))) { |
762 | | /* both sides stopped */ |
763 | 0 | must_stop = FD_POLL_IN | FD_POLL_OUT; |
764 | 0 | } |
765 | 0 | else if (unlikely(!fd_recv_active(fd) && (evts & (FD_EV_READY_R | FD_EV_SHUT_R | FD_EV_ERR_RW)))) { |
766 | | /* only send remains */ |
767 | 0 | must_stop = FD_POLL_IN; |
768 | 0 | } |
769 | 0 | else if (unlikely(!fd_send_active(fd) && (evts & (FD_EV_READY_W | FD_EV_SHUT_W | FD_EV_ERR_RW)))) { |
770 | | /* only recv remains */ |
771 | 0 | must_stop = FD_POLL_OUT; |
772 | 0 | } |
773 | |
|
774 | 0 | if (new_flags & (FD_POLL_IN | FD_POLL_HUP | FD_POLL_ERR)) |
775 | 0 | new_flags |= FD_EV_READY_R; |
776 | |
|
777 | 0 | if (new_flags & (FD_POLL_OUT | FD_POLL_ERR)) |
778 | 0 | new_flags |= FD_EV_READY_W; |
779 | |
|
780 | 0 | old = fdtab[fd].state; |
781 | 0 | new = (old & ~FD_POLL_UPDT_MASK) | new_flags; |
782 | |
|
783 | 0 | if (unlikely(locked)) { |
784 | | /* Locked FDs (those with more than 2 threads) are atomically updated */ |
785 | 0 | while (unlikely(new != old && !_HA_ATOMIC_CAS(&fdtab[fd].state, &old, new))) |
786 | 0 | new = (old & ~FD_POLL_UPDT_MASK) | new_flags; |
787 | 0 | } else { |
788 | 0 | if (new != old) |
789 | 0 | fdtab[fd].state = new; |
790 | 0 | } |
791 | |
|
792 | 0 | if (fdtab[fd].iocb && fd_active(fd)) { |
793 | 0 | fdtab[fd].iocb(fd); |
794 | 0 | } |
795 | | |
796 | | /* |
797 | | * We entered iocb with running set and with the valid tgid. |
798 | | * Since then, this is what could have happened: |
799 | | * - another thread tried to close the FD (e.g. timeout task from |
800 | | * another one that owns it). We still have running set, but not |
801 | | * tmask. We must call fd_clr_running() then _fd_delete_orphan() |
802 | | * if we were the last one. |
803 | | * |
804 | | * - the iocb tried to close the FD => bit no more present in running, |
805 | | * nothing to do. If it managed to close it, the poller's ->clo() |
806 | | * has already been called. |
807 | | * |
808 | | * - after we closed, the FD was reassigned to another thread in |
809 | | * another group => running not present, tgid differs, nothing to |
810 | | * do because if it got reassigned it indicates it was already |
811 | | * closed. |
812 | | * |
813 | | * There's no risk of takeover of the valid FD here during this period. |
814 | | * Also if we still have running, immediately after we release it, the |
815 | | * events above might instantly happen due to another thread taking |
816 | | * over. |
817 | | * |
818 | | * As such, the only cases where the FD is still relevant are: |
819 | | * - tgid still set and running still set (most common) |
820 | | * - tgid still valid but running cleared due to fd_delete(): we may |
821 | | * still need to stop polling otherwise we may keep it enabled |
822 | | * while waiting for other threads to close it. |
823 | | * And given that we may need to program a tentative update in case we |
824 | | * don't immediately close, it's easier to grab the tgid during the |
825 | | * whole check. |
826 | | */ |
827 | |
|
828 | 0 | if (!fd_grab_tgid(fd, tgid)) |
829 | 0 | return FD_UPDT_CLOSED; |
830 | | |
831 | 0 | tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask); |
832 | | |
833 | | /* another thread might have attempted to close this FD in the mean |
834 | | * time (e.g. timeout task) striking on a previous thread and closing. |
835 | | * This is detected by us being the last owners of a running_mask bit, |
836 | | * and the thread_mask being zero. At the moment we release the running |
837 | | * bit, a takeover may also happen, so in practice we check for our loss |
838 | | * of the thread_mask bitboth thread_mask and running_mask being 0 after |
839 | | * we remove ourselves last. There is no risk the FD gets reassigned |
840 | | * to a different group since it's not released until the real close() |
841 | | * in _fd_delete_orphan(). |
842 | | */ |
843 | 0 | if (fd_clr_running(fd) == ti->ltid_bit && !(tmask & ti->ltid_bit)) |
844 | 0 | goto closed_or_migrated; |
845 | | |
846 | | /* we had to stop this FD and it still must be stopped after the I/O |
847 | | * cb's changes, so let's program an update for this. |
848 | | */ |
849 | 0 | if (must_stop && !(fdtab[fd].update_mask & ti->ltid_bit)) { |
850 | 0 | if (((must_stop & FD_POLL_IN) && !fd_recv_active(fd)) || |
851 | 0 | ((must_stop & FD_POLL_OUT) && !fd_send_active(fd))) |
852 | 0 | if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid)) |
853 | 0 | fd_updt[fd_nbupdt++] = fd; |
854 | 0 | } |
855 | |
|
856 | 0 | fd_drop_tgid(fd); |
857 | 0 | return FD_UPDT_DONE; |
858 | | |
859 | 0 | closed_or_migrated: |
860 | | /* We only come here once we've last dropped running and the FD is |
861 | | * not for us as per !(tmask & tid_bit). It may imply we're |
862 | | * responsible for closing it. Otherwise it's just a migration. |
863 | | */ |
864 | 0 | if (HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) { |
865 | 0 | fd_drop_tgid(fd); |
866 | 0 | _fd_delete_orphan(fd); |
867 | 0 | return FD_UPDT_CLOSED; |
868 | 0 | } |
869 | | |
870 | | /* So we were alone, no close bit, at best the FD was migrated, at |
871 | | * worst it's in the process of being closed by another thread. We must |
872 | | * be ultra-careful as it can be re-inserted by yet another thread as |
873 | | * the result of socket() or accept(). Let's just tell the poller the |
874 | | * FD was lost. If it was closed it was already removed and this will |
875 | | * only cost an update for nothing. |
876 | | */ |
877 | | |
878 | 0 | do_update: |
879 | | /* The FD is not closed but we don't want the poller to wake up for |
880 | | * it anymore. |
881 | | */ |
882 | 0 | if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid)) |
883 | 0 | fd_updt[fd_nbupdt++] = fd; |
884 | |
|
885 | 0 | fd_drop_tgid(fd); |
886 | 0 | return FD_UPDT_MIGRATED; |
887 | 0 | } |
888 | | |
889 | | /* This is used by pollers at boot time to re-register desired events for |
890 | | * all FDs after new pollers have been created. It doesn't do much, it checks |
891 | | * that their thread group matches the one in argument, and that the thread |
892 | | * mask matches at least one of the bits in the mask, and if so, marks the FD |
893 | | * as updated. |
894 | | */ |
895 | | void fd_reregister_all(int tgrp, ulong mask) |
896 | 0 | { |
897 | 0 | int fd; |
898 | |
|
899 | 0 | for (fd = 0; fd < fd_highest; fd++) { |
900 | 0 | if (!fdtab[fd].owner) |
901 | 0 | continue; |
902 | | |
903 | | /* make sure we don't register other tgroups' FDs. We just |
904 | | * avoid needlessly taking the lock if not needed. |
905 | | */ |
906 | 0 | if (!(_HA_ATOMIC_LOAD(&fdtab[fd].thread_mask) & mask) || |
907 | 0 | !fd_grab_tgid(fd, tgrp)) |
908 | 0 | continue; // was not for us anyway |
909 | | |
910 | 0 | if (_HA_ATOMIC_LOAD(&fdtab[fd].thread_mask) & mask) |
911 | 0 | updt_fd_polling(fd); |
912 | 0 | fd_drop_tgid(fd); |
913 | 0 | } |
914 | 0 | } |
915 | | |
916 | | /* Tries to send <npfx> parts from <prefix> followed by <nmsg> parts from <msg> |
917 | | * optionally followed by a newline if <nl> is non-null, to file descriptor |
918 | | * <fd>. The message is sent atomically using writev(). It may be truncated to |
919 | | * <maxlen> bytes if <maxlen> is non-null. There is no distinction between the |
920 | | * two lists, it's just a convenience to help the caller prepend some prefixes |
921 | | * when necessary. It takes the fd's lock to make sure no other thread will |
922 | | * write to the same fd in parallel. Returns the number of bytes sent, or <=0 |
923 | | * on failure. A limit to 31 total non-empty segments is enforced. The caller |
924 | | * is responsible for taking care of making the fd non-blocking. |
925 | | */ |
926 | | ssize_t fd_write_frag_line(int fd, size_t maxlen, const struct ist pfx[], size_t npfx, const struct ist msg[], size_t nmsg, int nl) |
927 | 0 | { |
928 | 0 | struct iovec iovec[32]; |
929 | 0 | size_t sent = 0; |
930 | 0 | int vec = 0; |
931 | 0 | int attempts = 0; |
932 | |
|
933 | 0 | if (!maxlen) |
934 | 0 | maxlen = ~0; |
935 | | |
936 | | /* keep one char for a possible trailing '\n' in any case */ |
937 | 0 | maxlen--; |
938 | | |
939 | | /* make an iovec from the concatenation of all parts of the original |
940 | | * message. Skip empty fields and truncate the whole message to maxlen, |
941 | | * leaving one spare iovec for the '\n'. |
942 | | */ |
943 | 0 | while (vec < (sizeof(iovec) / sizeof(iovec[0]) - 1)) { |
944 | 0 | if (!npfx) { |
945 | 0 | pfx = msg; |
946 | 0 | npfx = nmsg; |
947 | 0 | nmsg = 0; |
948 | 0 | if (!npfx) |
949 | 0 | break; |
950 | 0 | } |
951 | | |
952 | 0 | iovec[vec].iov_base = pfx->ptr; |
953 | 0 | iovec[vec].iov_len = MIN(maxlen, pfx->len); |
954 | 0 | maxlen -= iovec[vec].iov_len; |
955 | 0 | if (iovec[vec].iov_len) |
956 | 0 | vec++; |
957 | 0 | pfx++; npfx--; |
958 | 0 | }; |
959 | |
|
960 | 0 | if (nl) { |
961 | 0 | iovec[vec].iov_base = "\n"; |
962 | 0 | iovec[vec].iov_len = 1; |
963 | 0 | vec++; |
964 | 0 | } |
965 | | |
966 | | /* make sure we never interleave writes and we never block. This means |
967 | | * we prefer to fail on collision than to block. But we don't want to |
968 | | * lose too many logs so we just perform a few lock attempts then give |
969 | | * up. |
970 | | */ |
971 | |
|
972 | 0 | while (HA_ATOMIC_BTS(&fdtab[fd].state, FD_EXCL_SYSCALL_BIT)) { |
973 | 0 | if (++attempts >= 200) { |
974 | | /* so that the caller knows the message couldn't be delivered */ |
975 | 0 | sent = -1; |
976 | 0 | errno = EAGAIN; |
977 | 0 | goto leave; |
978 | 0 | } |
979 | 0 | ha_thread_relax(); |
980 | 0 | } |
981 | | |
982 | 0 | if (unlikely(!(fdtab[fd].state & FD_INITIALIZED))) { |
983 | 0 | HA_ATOMIC_OR(&fdtab[fd].state, FD_INITIALIZED); |
984 | 0 | if (!isatty(fd)) |
985 | 0 | fd_set_nonblock(fd); |
986 | 0 | } |
987 | 0 | sent = writev(fd, iovec, vec); |
988 | 0 | HA_ATOMIC_BTR(&fdtab[fd].state, FD_EXCL_SYSCALL_BIT); |
989 | |
|
990 | 0 | leave: |
991 | | /* sent > 0 if the message was delivered */ |
992 | 0 | return sent; |
993 | 0 | } |
994 | | |
995 | | #if defined(USE_CLOSEFROM) |
996 | | void my_closefrom(int start) |
997 | | { |
998 | | closefrom(start); |
999 | | } |
1000 | | |
1001 | | #elif defined(USE_POLL) |
1002 | | /* This is a portable implementation of closefrom(). It closes all open file |
1003 | | * descriptors starting at <start> and above. It relies on the fact that poll() |
1004 | | * will return POLLNVAL for each invalid (hence close) file descriptor passed |
1005 | | * in argument in order to skip them. It acts with batches of FDs and will |
1006 | | * typically perform one poll() call per 1024 FDs so the overhead is low in |
1007 | | * case all FDs have to be closed. |
1008 | | */ |
1009 | | void my_closefrom(int start) |
1010 | 0 | { |
1011 | 0 | struct pollfd poll_events[1024]; |
1012 | 0 | struct rlimit limit; |
1013 | 0 | int nbfds, fd, ret, idx; |
1014 | 0 | int step, next; |
1015 | |
|
1016 | 0 | if (getrlimit(RLIMIT_NOFILE, &limit) == 0) |
1017 | 0 | step = nbfds = limit.rlim_cur; |
1018 | 0 | else |
1019 | 0 | step = nbfds = 0; |
1020 | |
|
1021 | 0 | if (nbfds <= 0) { |
1022 | | /* set safe limit */ |
1023 | 0 | nbfds = 1024; |
1024 | 0 | step = 256; |
1025 | 0 | } |
1026 | |
|
1027 | 0 | if (step > sizeof(poll_events) / sizeof(poll_events[0])) |
1028 | 0 | step = sizeof(poll_events) / sizeof(poll_events[0]); |
1029 | |
|
1030 | 0 | while (start < nbfds) { |
1031 | 0 | next = (start / step + 1) * step; |
1032 | |
|
1033 | 0 | for (fd = start; fd < next && fd < nbfds; fd++) { |
1034 | 0 | poll_events[fd - start].fd = fd; |
1035 | 0 | poll_events[fd - start].events = 0; |
1036 | 0 | } |
1037 | |
|
1038 | 0 | do { |
1039 | 0 | ret = poll(poll_events, fd - start, 0); |
1040 | 0 | if (ret >= 0) |
1041 | 0 | break; |
1042 | 0 | } while (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR || errno == ENOMEM); |
1043 | | |
1044 | | /* always check the whole range */ |
1045 | 0 | ret = fd - start; |
1046 | |
|
1047 | 0 | for (idx = 0; idx < ret; idx++) { |
1048 | 0 | if (poll_events[idx].revents & POLLNVAL) |
1049 | 0 | continue; /* already closed */ |
1050 | | |
1051 | 0 | fd = poll_events[idx].fd; |
1052 | 0 | close(fd); |
1053 | 0 | } |
1054 | 0 | start = next; |
1055 | 0 | } |
1056 | 0 | } |
1057 | | |
1058 | | #else // defined(USE_POLL) |
1059 | | |
1060 | | /* This is a portable implementation of closefrom(). It closes all open file |
1061 | | * descriptors starting at <start> and above. This is a naive version for use |
1062 | | * when the operating system provides no alternative. |
1063 | | */ |
1064 | | void my_closefrom(int start) |
1065 | | { |
1066 | | struct rlimit limit; |
1067 | | int nbfds; |
1068 | | |
1069 | | if (getrlimit(RLIMIT_NOFILE, &limit) == 0) |
1070 | | nbfds = limit.rlim_cur; |
1071 | | else |
1072 | | nbfds = 0; |
1073 | | |
1074 | | if (nbfds <= 0) |
1075 | | nbfds = 1024; /* safe limit */ |
1076 | | |
1077 | | while (start < nbfds) |
1078 | | close(start++); |
1079 | | } |
1080 | | #endif // defined(USE_POLL) |
1081 | | |
1082 | | |
1083 | | /* Computes the bounded poll() timeout based on the next expiration timer <next> |
1084 | | * by bounding it to MAX_DELAY_MS. <next> may equal TICK_ETERNITY. The pollers |
1085 | | * just needs to call this function right before polling to get their timeout |
1086 | | * value. Timeouts that are already expired (possibly due to a pending event) |
1087 | | * are accounted for in activity.poll_exp. |
1088 | | */ |
1089 | | int compute_poll_timeout(int next) |
1090 | 0 | { |
1091 | 0 | int wait_time; |
1092 | |
|
1093 | 0 | if (!tick_isset(next)) |
1094 | 0 | wait_time = MAX_DELAY_MS; |
1095 | 0 | else if (tick_is_expired(next, now_ms)) { |
1096 | 0 | activity[tid].poll_exp++; |
1097 | 0 | wait_time = 0; |
1098 | 0 | } |
1099 | 0 | else { |
1100 | 0 | wait_time = TICKS_TO_MS(tick_remain(now_ms, next)); |
1101 | 0 | if (wait_time > MAX_DELAY_MS) |
1102 | 0 | wait_time = MAX_DELAY_MS; |
1103 | 0 | } |
1104 | 0 | return wait_time; |
1105 | 0 | } |
1106 | | |
1107 | | /* Handle the return of the poller, which consists in calculating the idle |
1108 | | * time, saving a few clocks, marking the thread harmful again etc. All that |
1109 | | * is some boring stuff that all pollers have to do anyway. |
1110 | | */ |
1111 | | void fd_leaving_poll(int wait_time, int status) |
1112 | 0 | { |
1113 | 0 | clock_leaving_poll(wait_time, status); |
1114 | |
|
1115 | 0 | thread_harmless_end(); |
1116 | 0 | thread_idle_end(); |
1117 | |
|
1118 | 0 | _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_SLEEPING); |
1119 | 0 | } |
1120 | | |
1121 | | /* disable the specified poller */ |
1122 | | void disable_poller(const char *poller_name) |
1123 | 0 | { |
1124 | 0 | int p; |
1125 | |
|
1126 | 0 | for (p = 0; p < nbpollers; p++) |
1127 | 0 | if (strcmp(pollers[p].name, poller_name) == 0) |
1128 | 0 | pollers[p].pref = 0; |
1129 | 0 | } |
1130 | | |
1131 | | void poller_pipe_io_handler(int fd) |
1132 | 0 | { |
1133 | 0 | char buf[1024]; |
1134 | | /* Flush the pipe */ |
1135 | 0 | while (read(fd, buf, sizeof(buf)) > 0); |
1136 | 0 | fd_cant_recv(fd); |
1137 | 0 | } |
1138 | | |
1139 | | /* allocate the per-thread fd_updt thus needs to be called early after |
1140 | | * thread creation. |
1141 | | */ |
1142 | | static int alloc_pollers_per_thread() |
1143 | 0 | { |
1144 | 0 | fd_updt = calloc(global.maxsock, sizeof(*fd_updt)); |
1145 | 0 | vma_set_name_id(fd_updt, global.maxsock * sizeof(*fd_updt), "fd", "fd_updt", tid + 1); |
1146 | 0 | return fd_updt != NULL; |
1147 | 0 | } |
1148 | | |
1149 | | static int precreated_poller_pipes[MAX_THREADS][2]; |
1150 | | |
1151 | | /* |
1152 | | * Create the pipes before we create the threads, as they may not share |
1153 | | * their file descriptor tables. |
1154 | | */ |
1155 | | int fd_precreate_poller_pipes(void) |
1156 | 0 | { |
1157 | 0 | int i; |
1158 | |
|
1159 | 0 | for (i = 0; i < global.nbthread; i++) { |
1160 | 0 | if (pipe(precreated_poller_pipes[i]) < 0) |
1161 | 0 | return 0; |
1162 | 0 | } |
1163 | 0 | return 1; |
1164 | 0 | } |
1165 | | |
1166 | | /* Initialize the pollers per thread.*/ |
1167 | | static int init_pollers_per_thread() |
1168 | 0 | { |
1169 | 0 | int mypipe[2]; |
1170 | | |
1171 | | /* |
1172 | | * The master does not get to pre-create pipes, so do it |
1173 | | * now. |
1174 | | */ |
1175 | 0 | if (!master) { |
1176 | 0 | mypipe[0] = precreated_poller_pipes[tid][0]; |
1177 | 0 | mypipe[1] = precreated_poller_pipes[tid][1]; |
1178 | 0 | } |
1179 | 0 | else if (pipe(mypipe) < 0) |
1180 | 0 | return 0; |
1181 | | |
1182 | 0 | poller_rd_pipe = mypipe[0]; |
1183 | 0 | poller_wr_pipe[tid] = mypipe[1]; |
1184 | 0 | fd_set_nonblock(poller_rd_pipe); |
1185 | 0 | fd_insert(poller_rd_pipe, poller_pipe_io_handler, poller_pipe_io_handler, tgid, ti->ltid_bit); |
1186 | 0 | fd_insert(poller_wr_pipe[tid], poller_pipe_io_handler, poller_pipe_io_handler, tgid, ti->ltid_bit); |
1187 | 0 | fd_want_recv(poller_rd_pipe); |
1188 | 0 | fd_stop_both(poller_wr_pipe[tid]); |
1189 | 0 | return 1; |
1190 | 0 | } |
1191 | | |
1192 | | /* Deinitialize the pollers per thread */ |
1193 | | static void deinit_pollers_per_thread() |
1194 | 0 | { |
1195 | | /* rd and wr are init at the same place, but only rd is init to -1, so |
1196 | | we rely to rd to close. */ |
1197 | 0 | if (poller_rd_pipe > -1) { |
1198 | 0 | fd_delete(poller_rd_pipe); |
1199 | 0 | poller_rd_pipe = -1; |
1200 | 0 | fd_delete(poller_wr_pipe[tid]); |
1201 | 0 | poller_wr_pipe[tid] = -1; |
1202 | 0 | } |
1203 | 0 | } |
1204 | | |
1205 | | /* Release the pollers per thread, to be called late */ |
1206 | | static void free_pollers_per_thread() |
1207 | 0 | { |
1208 | 0 | fd_nbupdt = 0; |
1209 | 0 | ha_free(&fd_updt); |
1210 | 0 | } |
1211 | | |
1212 | | /* |
1213 | | * Initialize the pollers till the best one is found. |
1214 | | * If none works, returns 0, otherwise 1. |
1215 | | */ |
1216 | | int init_pollers() |
1217 | 0 | { |
1218 | 0 | int p; |
1219 | 0 | struct poller *bp; |
1220 | 0 | int nbfdtab; |
1221 | | |
1222 | |
|
1223 | 0 | if (global.tune.options & GTUNE_NO_TG_FD_SHARING) { |
1224 | 0 | nbfdtab = global.nbtgroups; |
1225 | 0 | } else |
1226 | 0 | nbfdtab = 1; |
1227 | | /* always provide an aligned fdtab */ |
1228 | 0 | if ((_fdtab = ha_aligned_zalloc(64, array_size_or_fail(global.maxsock, nbfdtab * sizeof(*fdtab)))) == NULL) { |
1229 | 0 | ha_alert("Not enough memory to allocate %d entries for fdtab!\n", global.maxsock); |
1230 | 0 | goto fail_tab; |
1231 | 0 | } |
1232 | | |
1233 | 0 | for (p = 0; p < global.nbtgroups; p++) { |
1234 | 0 | if (global.tune.options & GTUNE_NO_TG_FD_SHARING) { |
1235 | 0 | ha_tgroup_ctx[p].fdtab = (void *)((char *)(void *)_fdtab + p * global.maxsock * sizeof(*fdtab)); |
1236 | 0 | } else |
1237 | 0 | ha_tgroup_ctx[p].fdtab = _fdtab; |
1238 | 0 | } |
1239 | | /* |
1240 | | * Immediately set the fdtab for our thread, as we'll need it soon |
1241 | | */ |
1242 | 0 | fdtab = ha_tgroup_ctx[0].fdtab; |
1243 | 0 | vma_set_name(_fdtab, global.maxsock * sizeof(*fdtab) * nbfdtab, "fd", "fdtab"); |
1244 | |
|
1245 | 0 | if ((_polled_mask = calloc(global.maxsock, nbfdtab * sizeof(*polled_mask))) == NULL) { |
1246 | 0 | ha_alert("Not enough memory to allocate %d entries for polled_mask!\n", global.maxsock); |
1247 | 0 | goto fail_polledmask; |
1248 | 0 | } |
1249 | | |
1250 | | /* |
1251 | | * As for fdtab, get on polled_mask per thread group, if needed. |
1252 | | */ |
1253 | 0 | for (p = 0; p < global.nbtgroups; p++) { |
1254 | 0 | if (global.tune.options & GTUNE_NO_TG_FD_SHARING) |
1255 | 0 | ha_tgroup_ctx[p].polled_mask = _polled_mask + p * global.maxsock; |
1256 | 0 | else |
1257 | 0 | ha_tgroup_ctx[p].polled_mask = _polled_mask; |
1258 | 0 | } |
1259 | | /* immediately set it for the boot thread as well */ |
1260 | 0 | polled_mask = ha_tgroup_ctx[0].polled_mask; |
1261 | 0 | vma_set_name(_polled_mask, global.maxsock * sizeof(*polled_mask) * nbfdtab, "fd", "polled_mask"); |
1262 | |
|
1263 | 0 | for (p = 0; p < MAX_TGROUPS; p++) |
1264 | 0 | update_list[p].first = update_list[p].last = -1; |
1265 | |
|
1266 | 0 | for (p = 0; p < global.maxsock; p++) { |
1267 | | /* Mark the fd as out of the fd cache */ |
1268 | 0 | fdtab[p].update.next = -3; |
1269 | 0 | } |
1270 | |
|
1271 | 0 | do { |
1272 | 0 | bp = NULL; |
1273 | 0 | redo: |
1274 | 0 | for (p = 0; p < nbpollers; p++) |
1275 | 0 | if (!bp || (pollers[p].pref > bp->pref)) |
1276 | 0 | bp = &pollers[p]; |
1277 | |
|
1278 | 0 | if (!bp || bp->pref == 0) |
1279 | 0 | break; |
1280 | | |
1281 | 0 | if ((global.tune.options & GTUNE_NO_TG_FD_SHARING) && |
1282 | 0 | !(bp->flags & HAP_POLL_F_NO_FD_SHARING)) { |
1283 | 0 | ha_alert("Can't use poller '%s' with per-thread-group FD tables\n", bp->name); |
1284 | 0 | bp->pref = 0; |
1285 | 0 | goto redo; |
1286 | 0 | } |
1287 | | |
1288 | 0 | if (bp->init(bp)) { |
1289 | 0 | memcpy(&cur_poller, bp, sizeof(*bp)); |
1290 | 0 | return 1; |
1291 | 0 | } |
1292 | 0 | } while (!bp || bp->pref == 0); |
1293 | | |
1294 | 0 | fail_info: |
1295 | 0 | free(_polled_mask); |
1296 | 0 | _polled_mask = polled_mask = NULL; |
1297 | 0 | fail_polledmask: |
1298 | 0 | ha_aligned_free(_fdtab); |
1299 | 0 | _fdtab = fdtab = NULL; |
1300 | 0 | fail_tab: |
1301 | 0 | return 0; |
1302 | 0 | } |
1303 | | |
1304 | | /* |
1305 | | * Deinitialize the pollers. |
1306 | | */ |
1307 | 0 | void deinit_pollers() { |
1308 | |
|
1309 | 0 | struct poller *bp; |
1310 | 0 | int p; |
1311 | |
|
1312 | 0 | for (p = 0; p < nbpollers; p++) { |
1313 | 0 | bp = &pollers[p]; |
1314 | |
|
1315 | 0 | if (bp && bp->pref) |
1316 | 0 | bp->term(bp); |
1317 | 0 | } |
1318 | | |
1319 | | /* free the backing areas, not the (possibly per-group) thread-local |
1320 | | * pointers which may not point to the start of the allocations. |
1321 | | */ |
1322 | 0 | ha_aligned_free(_fdtab); |
1323 | 0 | _fdtab = fdtab = NULL; |
1324 | 0 | ha_free(&_polled_mask); |
1325 | 0 | polled_mask = NULL; |
1326 | 0 | } |
1327 | | |
1328 | | /* |
1329 | | * Lists the known pollers on <out>. |
1330 | | * Should be performed only before initialization. |
1331 | | */ |
1332 | | int list_pollers(FILE *out) |
1333 | 0 | { |
1334 | 0 | int p; |
1335 | 0 | int last, next; |
1336 | 0 | int usable; |
1337 | 0 | struct poller *bp; |
1338 | |
|
1339 | 0 | fprintf(out, "Available polling systems :\n"); |
1340 | |
|
1341 | 0 | usable = 0; |
1342 | 0 | bp = NULL; |
1343 | 0 | last = next = -1; |
1344 | 0 | while (1) { |
1345 | 0 | for (p = 0; p < nbpollers; p++) { |
1346 | 0 | if ((next < 0 || pollers[p].pref > next) |
1347 | 0 | && (last < 0 || pollers[p].pref < last)) { |
1348 | 0 | next = pollers[p].pref; |
1349 | 0 | if (!bp || (pollers[p].pref > bp->pref)) |
1350 | 0 | bp = &pollers[p]; |
1351 | 0 | } |
1352 | 0 | } |
1353 | |
|
1354 | 0 | if (next == -1) |
1355 | 0 | break; |
1356 | | |
1357 | 0 | for (p = 0; p < nbpollers; p++) { |
1358 | 0 | if (pollers[p].pref == next) { |
1359 | 0 | fprintf(out, " %10s : ", pollers[p].name); |
1360 | 0 | if (pollers[p].pref == 0) |
1361 | 0 | fprintf(out, "disabled, "); |
1362 | 0 | else |
1363 | 0 | fprintf(out, "pref=%3d, ", pollers[p].pref); |
1364 | 0 | if (pollers[p].test(&pollers[p])) { |
1365 | 0 | fprintf(out, " test result OK"); |
1366 | 0 | if (next > 0) |
1367 | 0 | usable++; |
1368 | 0 | } else { |
1369 | 0 | fprintf(out, " test result FAILED"); |
1370 | 0 | if (bp == &pollers[p]) |
1371 | 0 | bp = NULL; |
1372 | 0 | } |
1373 | 0 | fprintf(out, "\n"); |
1374 | 0 | } |
1375 | 0 | } |
1376 | 0 | last = next; |
1377 | 0 | next = -1; |
1378 | 0 | }; |
1379 | 0 | fprintf(out, "Total: %d (%d usable), will use %s.\n", nbpollers, usable, bp ? bp->name : "none"); |
1380 | 0 | return 0; |
1381 | 0 | } |
1382 | | |
1383 | | /* |
1384 | | * Some pollers may lose their connection after a fork(). It may be necessary |
1385 | | * to create initialize part of them again. Returns 0 in case of failure, |
1386 | | * otherwise 1. The fork() function may be NULL if unused. In case of error, |
1387 | | * the the current poller is destroyed and the caller is responsible for trying |
1388 | | * another one by calling init_pollers() again. |
1389 | | */ |
1390 | | int fork_poller() |
1391 | 0 | { |
1392 | 0 | int fd; |
1393 | 0 | for (fd = 0; fd < fd_highest; fd++) { |
1394 | 0 | if (fdtab[fd].owner) { |
1395 | 0 | HA_ATOMIC_OR(&fdtab[fd].state, FD_CLONED); |
1396 | 0 | } |
1397 | 0 | } |
1398 | |
|
1399 | 0 | if (cur_poller.fork) { |
1400 | 0 | if (cur_poller.fork(&cur_poller)) |
1401 | 0 | return 1; |
1402 | 0 | cur_poller.term(&cur_poller); |
1403 | 0 | return 0; |
1404 | 0 | } |
1405 | 0 | return 1; |
1406 | 0 | } |
1407 | | |
1408 | | /* config parser for global "tune.fd.edge-triggered", accepts "on" or "off" */ |
1409 | | static int cfg_parse_tune_fd_edge_triggered(char **args, int section_type, struct proxy *curpx, |
1410 | | const struct proxy *defpx, const char *file, int line, |
1411 | | char **err) |
1412 | 0 | { |
1413 | 0 | if (too_many_args(1, args, err, NULL)) |
1414 | 0 | return -1; |
1415 | | |
1416 | 0 | if (strcmp(args[1], "on") == 0) |
1417 | 0 | global.tune.options |= GTUNE_FD_ET; |
1418 | 0 | else if (strcmp(args[1], "off") == 0) |
1419 | 0 | global.tune.options &= ~GTUNE_FD_ET; |
1420 | 0 | else { |
1421 | 0 | memprintf(err, "'%s' expects either 'on' or 'off' but got '%s'.", args[0], args[1]); |
1422 | 0 | return -1; |
1423 | 0 | } |
1424 | 0 | return 0; |
1425 | 0 | } |
1426 | | |
1427 | | /* config keyword parsers */ |
1428 | | static struct cfg_kw_list cfg_kws = {ILH, { |
1429 | | { CFG_GLOBAL, "tune.fd.edge-triggered", cfg_parse_tune_fd_edge_triggered, KWF_EXPERIMENTAL }, |
1430 | | { 0, NULL, NULL } |
1431 | | }}; |
1432 | | |
1433 | | INITCALL1(STG_REGISTER, cfg_register_keywords, &cfg_kws); |
1434 | | |
1435 | | REGISTER_PER_THREAD_ALLOC(alloc_pollers_per_thread); |
1436 | | REGISTER_PER_THREAD_INIT(init_pollers_per_thread); |
1437 | | REGISTER_PER_THREAD_DEINIT(deinit_pollers_per_thread); |
1438 | | REGISTER_PER_THREAD_FREE(free_pollers_per_thread); |
1439 | | |
1440 | | /* |
1441 | | * Local variables: |
1442 | | * c-indent-level: 8 |
1443 | | * c-basic-offset: 8 |
1444 | | * End: |
1445 | | */ |