diff options
Diffstat (limited to 'src/fd.c')
| -rw-r--r-- | src/fd.c | 1348 |
1 files changed, 1348 insertions, 0 deletions
diff --git a/src/fd.c b/src/fd.c new file mode 100644 index 0000000..9d34315 --- /dev/null +++ b/src/fd.c @@ -0,0 +1,1348 @@ +/* + * File descriptors management functions. + * + * Copyright 2000-2014 Willy Tarreau <w@1wt.eu> + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version + * 2 of the License, or (at your option) any later version. + * + * There is no direct link between the FD and the updates list. There is only a + * bit in the fdtab[] to indicate than a file descriptor is already present in + * the updates list. Once an fd is present in the updates list, it will have to + * be considered even if its changes are reverted in the middle or if the fd is + * replaced. + * + * The event state for an FD, as found in fdtab[].state, is maintained for each + * direction. The state field is built this way, with R bits in the low nibble + * and W bits in the high nibble for ease of access and debugging : + * + * 7 6 5 4 3 2 1 0 + * [ 0 | 0 | RW | AW | 0 | 0 | RR | AR ] + * + * A* = active *R = read + * R* = ready *W = write + * + * An FD is marked "active" when there is a desire to use it. + * An FD is marked "ready" when it has not faced a new EAGAIN since last wake-up + * (it is a cache of the last EAGAIN regardless of polling changes). Each poller + * has its own "polled" state for the same fd, as stored in the polled_mask. + * + * We have 4 possible states for each direction based on these 2 flags : + * + * +---+---+----------+---------------------------------------------+ + * | R | A | State | Description | + * +---+---+----------+---------------------------------------------+ + * | 0 | 0 | DISABLED | No activity desired, not ready. | + * | 0 | 1 | ACTIVE | Activity desired. | + * | 1 | 0 | STOPPED | End of activity. | + * | 1 | 1 | READY | Activity desired and reported. | + * +---+---+----------+---------------------------------------------+ + * + * The transitions are pretty simple : + * - fd_want_*() : set flag A + * - fd_stop_*() : clear flag A + * - fd_cant_*() : clear flag R (when facing EAGAIN) + * - fd_may_*() : set flag R (upon return from poll()) + * + * Each poller then computes its own polled state : + * if (A) { if (!R) P := 1 } else { P := 0 } + * + * The state transitions look like the diagram below. + * + * may +----------+ + * ,----| DISABLED | (READY=0, ACTIVE=0) + * | +----------+ + * | want | ^ + * | | | + * | v | stop + * | +----------+ + * | | ACTIVE | (READY=0, ACTIVE=1) + * | +----------+ + * | | ^ + * | may | | + * | v | EAGAIN (can't) + * | +--------+ + * | | READY | (READY=1, ACTIVE=1) + * | +--------+ + * | stop | ^ + * | | | + * | v | want + * | +---------+ + * `--->| STOPPED | (READY=1, ACTIVE=0) + * +---------+ + */ + +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <fcntl.h> +#include <sys/types.h> +#include <sys/resource.h> +#include <sys/uio.h> + +#if defined(USE_POLL) +#include <poll.h> +#include <errno.h> +#endif + +#include <haproxy/api.h> +#include <haproxy/activity.h> +#include <haproxy/cfgparse.h> +#include <haproxy/fd.h> +#include <haproxy/global.h> +#include <haproxy/log.h> +#include <haproxy/port_range.h> +#include <haproxy/ticks.h> +#include <haproxy/tools.h> + + +struct fdtab *fdtab __read_mostly = NULL; /* array of all the file descriptors */ +struct polled_mask *polled_mask __read_mostly = NULL; /* Array for the polled_mask of each fd */ +struct fdinfo *fdinfo __read_mostly = NULL; /* less-often used infos for file descriptors */ +int totalconn; /* total # of terminated sessions */ +int actconn; /* # of active sessions */ + +struct poller pollers[MAX_POLLERS] __read_mostly; +struct poller cur_poller __read_mostly; +int nbpollers = 0; + +volatile struct fdlist update_list[MAX_TGROUPS]; // Global update list + +THREAD_LOCAL int *fd_updt = NULL; // FD updates list +THREAD_LOCAL int fd_nbupdt = 0; // number of updates in the list +THREAD_LOCAL int poller_rd_pipe = -1; // Pipe to wake the thread +int poller_wr_pipe[MAX_THREADS] __read_mostly; // Pipe to wake the threads + +volatile int ha_used_fds = 0; // Number of FD we're currently using +static struct fdtab *fdtab_addr; /* address of the allocated area containing fdtab */ + +/* adds fd <fd> to fd list <list> if it was not yet in it */ +void fd_add_to_fd_list(volatile struct fdlist *list, int fd) +{ + int next; + int new; + int old; + int last; + +redo_next: + next = HA_ATOMIC_LOAD(&fdtab[fd].update.next); + /* Check that we're not already in the cache, and if not, lock us. */ + if (next > -2) + goto done; + if (next == -2) + goto redo_next; + if (!_HA_ATOMIC_CAS(&fdtab[fd].update.next, &next, -2)) + goto redo_next; + __ha_barrier_atomic_store(); + + new = fd; +redo_last: + /* First, insert in the linked list */ + last = list->last; + old = -1; + + fdtab[fd].update.prev = -2; + /* Make sure the "prev" store is visible before we update the last entry */ + __ha_barrier_store(); + + if (unlikely(last == -1)) { + /* list is empty, try to add ourselves alone so that list->last=fd */ + if (unlikely(!_HA_ATOMIC_CAS(&list->last, &old, new))) + goto redo_last; + + /* list->first was necessary -1, we're guaranteed to be alone here */ + list->first = fd; + } else { + /* adding ourselves past the last element + * The CAS will only succeed if its next is -1, + * which means it's in the cache, and the last element. + */ + if (unlikely(!_HA_ATOMIC_CAS(&fdtab[last].update.next, &old, new))) + goto redo_last; + + /* Then, update the last entry */ + list->last = fd; + } + __ha_barrier_store(); + /* since we're alone at the end of the list and still locked(-2), + * we know no one tried to add past us. Mark the end of list. + */ + fdtab[fd].update.prev = last; + fdtab[fd].update.next = -1; + __ha_barrier_store(); +done: + return; +} + +/* removes fd <fd> from fd list <list> */ +void fd_rm_from_fd_list(volatile struct fdlist *list, int fd) +{ +#if defined(HA_HAVE_CAS_DW) || defined(HA_CAS_IS_8B) + volatile union { + struct fdlist_entry ent; + uint64_t u64; + uint32_t u32[2]; + } cur_list, next_list; +#endif + int old; + int new = -2; + int prev; + int next; + int last; +lock_self: +#if (defined(HA_CAS_IS_8B) || defined(HA_HAVE_CAS_DW)) + next_list.ent.next = next_list.ent.prev = -2; + cur_list.ent = *(volatile typeof(fdtab->update)*)&fdtab[fd].update; + /* First, attempt to lock our own entries */ + do { + /* The FD is not in the FD cache, give up */ + if (unlikely(cur_list.ent.next <= -3)) + return; + if (unlikely(cur_list.ent.prev == -2 || cur_list.ent.next == -2)) + goto lock_self; + } while ( +#ifdef HA_CAS_IS_8B + unlikely(!_HA_ATOMIC_CAS(((uint64_t *)&fdtab[fd].update), (uint64_t *)&cur_list.u64, next_list.u64)) +#else + unlikely(!_HA_ATOMIC_DWCAS(((long *)&fdtab[fd].update), (uint32_t *)&cur_list.u32, (const uint32_t *)&next_list.u32)) +#endif + ); + next = cur_list.ent.next; + prev = cur_list.ent.prev; + +#else +lock_self_next: + next = HA_ATOMIC_LOAD(&fdtab[fd].update.next); + if (next == -2) + goto lock_self_next; + if (next <= -3) + goto done; + if (unlikely(!_HA_ATOMIC_CAS(&fdtab[fd].update.next, &next, -2))) + goto lock_self_next; +lock_self_prev: + prev = HA_ATOMIC_LOAD(&fdtab[fd].update.prev); + if (prev == -2) + goto lock_self_prev; + if (unlikely(!_HA_ATOMIC_CAS(&fdtab[fd].update.prev, &prev, -2))) + goto lock_self_prev; +#endif + __ha_barrier_atomic_store(); + + /* Now, lock the entries of our neighbours */ + if (likely(prev != -1)) { +redo_prev: + old = fd; + + if (unlikely(!_HA_ATOMIC_CAS(&fdtab[prev].update.next, &old, new))) { + if (unlikely(old == -2)) { + /* Neighbour already locked, give up and + * retry again once he's done + */ + fdtab[fd].update.prev = prev; + __ha_barrier_store(); + fdtab[fd].update.next = next; + __ha_barrier_store(); + goto lock_self; + } + goto redo_prev; + } + } + if (likely(next != -1)) { +redo_next: + old = fd; + if (unlikely(!_HA_ATOMIC_CAS(&fdtab[next].update.prev, &old, new))) { + if (unlikely(old == -2)) { + /* Neighbour already locked, give up and + * retry again once he's done + */ + if (prev != -1) { + fdtab[prev].update.next = fd; + __ha_barrier_store(); + } + fdtab[fd].update.prev = prev; + __ha_barrier_store(); + fdtab[fd].update.next = next; + __ha_barrier_store(); + goto lock_self; + } + goto redo_next; + } + } + if (list->first == fd) + list->first = next; + __ha_barrier_store(); + last = list->last; + while (unlikely(last == fd && (!_HA_ATOMIC_CAS(&list->last, &last, prev)))) + __ha_compiler_barrier(); + /* Make sure we let other threads know we're no longer in cache, + * before releasing our neighbours. + */ + __ha_barrier_store(); + if (likely(prev != -1)) + fdtab[prev].update.next = next; + __ha_barrier_store(); + if (likely(next != -1)) + fdtab[next].update.prev = prev; + __ha_barrier_store(); + /* Ok, now we're out of the fd cache */ + fdtab[fd].update.next = -(next + 4); + __ha_barrier_store(); +done: + return; +} + +/* deletes the FD once nobody uses it anymore, as detected by the caller by its + * thread_mask being zero and its running mask turning to zero. There is no + * protection against concurrent accesses, it's up to the caller to make sure + * only the last thread will call it. If called under isolation, it is safe to + * call this from another group than the FD's. This is only for internal use, + * please use fd_delete() instead. + */ +void _fd_delete_orphan(int fd) +{ + int tgrp = fd_tgid(fd); + uint fd_disown; + + fd_disown = fdtab[fd].state & FD_DISOWN; + if (fdtab[fd].state & FD_LINGER_RISK) { + /* this is generally set when connecting to servers */ + DISGUISE(setsockopt(fd, SOL_SOCKET, SO_LINGER, + (struct linger *) &nolinger, sizeof(struct linger))); + } + + /* It's expected that a close() will result in the FD disappearing from + * pollers, but some pollers may have some internal bookkeeping to be + * done prior to the call (e.g. remove references from internal tables). + */ + if (cur_poller.clo) + cur_poller.clo(fd); + + /* now we're about to reset some of this FD's fields. We don't want + * anyone to grab it anymore and we need to make sure those which could + * possibly have stumbled upon it right now are leaving before we + * proceed. This is done in two steps. First we reset the tgid so that + * fd_take_tgid() and fd_grab_tgid() fail, then we wait for existing + * ref counts to drop. Past this point we're alone dealing with the + * FD's thead/running/update/polled masks. + */ + fd_reset_tgid(fd); + + while (_HA_ATOMIC_LOAD(&fdtab[fd].refc_tgid) != 0) // refc==0 ? + __ha_cpu_relax(); + + /* we don't want this FD anymore in the global list */ + fd_rm_from_fd_list(&update_list[tgrp - 1], fd); + + /* no more updates on this FD are relevant anymore */ + HA_ATOMIC_STORE(&fdtab[fd].update_mask, 0); + if (fd_nbupdt > 0 && fd_updt[fd_nbupdt - 1] == fd) + fd_nbupdt--; + + port_range_release_port(fdinfo[fd].port_range, fdinfo[fd].local_port); + polled_mask[fd].poll_recv = polled_mask[fd].poll_send = 0; + + fdtab[fd].state = 0; + +#ifdef DEBUG_FD + fdtab[fd].event_count = 0; +#endif + fdinfo[fd].port_range = NULL; + fdtab[fd].owner = NULL; + + /* perform the close() call last as it's what unlocks the instant reuse + * of this FD by any other thread. + */ + if (!fd_disown) + close(fd); + _HA_ATOMIC_DEC(&ha_used_fds); +} + +/* Deletes an FD from the fdsets. The file descriptor is also closed, possibly + * asynchronously. It is safe to call it from another thread from the same + * group as the FD's or from a thread from a different group. However if called + * from a thread from another group, there is an extra cost involved because + * the operation is performed under thread isolation, so doing so must be + * reserved for ultra-rare cases (e.g. stopping a listener). + */ +void fd_delete(int fd) +{ + /* This must never happen and would definitely indicate a bug, in + * addition to overwriting some unexpected memory areas. + */ + BUG_ON(fd < 0 || fd >= global.maxsock); + + /* NOTE: The master when going into reexec mode re-closes all FDs after + * they were already dispatched. But we know we didn't start the polling + * threads so we can still close them. The masks will probably not match + * however so we force the value and erase the refcount if any. + */ + if (unlikely(global.mode & MODE_STARTING)) + fdtab[fd].refc_tgid = ti->tgid; + + /* the tgid cannot change before a complete close so we should never + * face the situation where we try to close an fd that was reassigned. + * However there is one corner case where this happens, it's when an + * attempt to pause a listener fails (e.g. abns), leaving the listener + * in fault state and it is forcefully stopped. This needs to be done + * under isolation, and it's quite rare (i.e. once per such FD per + * process). Since we'll be isolated we can clear the thread mask and + * close the FD ourselves. + */ + if (unlikely(fd_tgid(fd) != ti->tgid)) { + int must_isolate = !thread_isolated() && !(global.mode & MODE_STOPPING); + + if (must_isolate) + thread_isolate(); + + HA_ATOMIC_STORE(&fdtab[fd].thread_mask, 0); + HA_ATOMIC_STORE(&fdtab[fd].running_mask, 0); + _fd_delete_orphan(fd); + + if (must_isolate) + thread_release(); + return; + } + + /* we must postpone removal of an FD that may currently be in use + * by another thread. This can happen in the following two situations: + * - after a takeover, the owning thread closes the connection but + * the previous one just woke up from the poller and entered + * the FD handler iocb. That thread holds an entry in running_mask + * and requires removal protection. + * - multiple threads are accepting connections on a listener, and + * one of them (or even an separate one) decides to unbind the + * listener under the listener's lock while other ones still hold + * the running bit. + * In both situations the FD is marked as unused (thread_mask = 0) and + * will not take new bits in its running_mask so we have the guarantee + * that the last thread eliminating running_mask is the one allowed to + * safely delete the FD. Most of the time it will be the current thread. + * We still need to set and check the one-shot flag FD_MUST_CLOSE + * to take care of the rare cases where a thread wakes up on late I/O + * before the thread_mask is zero, and sets its bit in the running_mask + * just after the current thread finishes clearing its own bit, hence + * the two threads see themselves as last ones (which they really are). + */ + + HA_ATOMIC_OR(&fdtab[fd].running_mask, ti->ltid_bit); + HA_ATOMIC_OR(&fdtab[fd].state, FD_MUST_CLOSE); + HA_ATOMIC_STORE(&fdtab[fd].thread_mask, 0); + if (fd_clr_running(fd) == ti->ltid_bit) { + if (HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) { + _fd_delete_orphan(fd); + } + } +} + +/* makes the new fd non-blocking and clears all other O_* flags; this is meant + * to be used on new FDs. Returns -1 on failure. The result is disguised at the + * end because some callers need to be able to ignore it regardless of the libc + * attributes. + */ +int fd_set_nonblock(int fd) +{ + int ret = fcntl(fd, F_SETFL, O_NONBLOCK); + + return DISGUISE(ret); +} + +/* sets the close-on-exec flag on fd; returns -1 on failure. The result is + * disguised at the end because some callers need to be able to ignore it + * regardless of the libc attributes. + */ +int fd_set_cloexec(int fd) +{ + int flags, ret; + + flags = fcntl(fd, F_GETFD); + flags |= FD_CLOEXEC; + ret = fcntl(fd, F_SETFD, flags); + return DISGUISE(ret); +} + +/* Migrate a FD to a new thread <new_tid>. It is explicitly permitted to + * migrate to another thread group, the function takes the necessary locking + * for this. It is even permitted to migrate from a foreign group to another, + * but the calling thread must be certain that the FD is not about to close + * when doing so, reason why it is highly recommended that only one of the + * FD's owners performs this operation. The polling is completely disabled. + * The operation never fails. + */ +void fd_migrate_on(int fd, uint new_tid) +{ + struct thread_info *new_ti = &ha_thread_info[new_tid]; + + /* we must be alone to work on this idle FD. If not, it means that its + * poller is currently waking up and is about to use it, likely to + * close it on shut/error, but maybe also to process any unexpectedly + * pending data. It's also possible that the FD was closed and + * reassigned to another thread group, so let's be careful. + */ + fd_lock_tgid(fd, new_ti->tgid); + + /* now we have exclusive access to it. From now FD belongs to tid_bit + * for this tgid. + */ + HA_ATOMIC_STORE(&fdtab[fd].thread_mask, new_ti->ltid_bit); + + /* Make sure the FD doesn't have the active bit. It is possible that + * the fd is polled by the thread that used to own it, the new thread + * is supposed to call subscribe() later, to activate polling. + */ + fd_stop_both(fd); + + /* we're done with it. As soon as we unlock it, other threads from the + * target group can manipulate it. However it may only disappear once + * we drop the reference. + */ + fd_unlock_tgid(fd); + fd_drop_tgid(fd); +} + +/* + * Take over a FD belonging to another thread. + * unexpected_conn is the expected owner of the fd. + * Returns 0 on success, and -1 on failure. + */ +int fd_takeover(int fd, void *expected_owner) +{ + unsigned long old; + + /* protect ourself against a delete then an insert for the same fd, + * if it happens, then the owner will no longer be the expected + * connection. + */ + if (fdtab[fd].owner != expected_owner) + return -1; + + /* we must be alone to work on this idle FD. If not, it means that its + * poller is currently waking up and is about to use it, likely to + * close it on shut/error, but maybe also to process any unexpectedly + * pending data. It's also possible that the FD was closed and + * reassigned to another thread group, so let's be careful. + */ + if (unlikely(!fd_grab_tgid(fd, ti->tgid))) + return -1; + + old = 0; + if (!HA_ATOMIC_CAS(&fdtab[fd].running_mask, &old, ti->ltid_bit)) { + fd_drop_tgid(fd); + return -1; + } + + /* success, from now on it's ours */ + HA_ATOMIC_STORE(&fdtab[fd].thread_mask, ti->ltid_bit); + + /* Make sure the FD doesn't have the active bit. It is possible that + * the fd is polled by the thread that used to own it, the new thread + * is supposed to call subscribe() later, to activate polling. + */ + fd_stop_recv(fd); + + /* we're done with it */ + HA_ATOMIC_AND(&fdtab[fd].running_mask, ~ti->ltid_bit); + + /* no more changes planned */ + fd_drop_tgid(fd); + return 0; +} + +void updt_fd_polling(const int fd) +{ + uint tgrp = fd_take_tgid(fd); + + /* closed ? may happen */ + if (!tgrp) + return; + + if (unlikely(tgrp != tgid && tgrp <= MAX_TGROUPS)) { + /* Hmmm delivered an update for another group... That may + * happen on suspend/resume of a listener for example when + * the FD was not even marked for running. Let's broadcast + * the update. + */ + unsigned long update_mask = fdtab[fd].update_mask; + int thr; + + while (!_HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask, + _HA_ATOMIC_LOAD(&ha_tgroup_info[tgrp - 1].threads_enabled))) + __ha_cpu_relax(); + + fd_add_to_fd_list(&update_list[tgrp - 1], fd); + + thr = one_among_mask(fdtab[fd].thread_mask & ha_tgroup_info[tgrp - 1].threads_enabled, + statistical_prng_range(ha_tgroup_info[tgrp - 1].count)); + thr += ha_tgroup_info[tgrp - 1].base; + wake_thread(thr); + + fd_drop_tgid(fd); + return; + } + + fd_drop_tgid(fd); + + if (tg->threads_enabled == 1UL || (fdtab[fd].thread_mask & tg->threads_enabled) == ti->ltid_bit) { + if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid)) + return; + + fd_updt[fd_nbupdt++] = fd; + } else { + unsigned long update_mask = fdtab[fd].update_mask; + do { + if (update_mask == fdtab[fd].thread_mask) // FIXME: this works only on thread-groups 1 + return; + } while (!_HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask, fdtab[fd].thread_mask)); + + fd_add_to_fd_list(&update_list[tgid - 1], fd); + + if (fd_active(fd) && !(fdtab[fd].thread_mask & ti->ltid_bit)) { + /* we need to wake up another thread to handle it immediately, any will fit, + * so let's pick a random one so that it doesn't always end up on the same. + */ + int thr = one_among_mask(fdtab[fd].thread_mask & tg->threads_enabled, + statistical_prng_range(tg->count)); + thr += tg->base; + wake_thread(thr); + } + } +} + +/* Update events seen for FD <fd> and its state if needed. This should be + * called by the poller, passing FD_EV_*_{R,W,RW} in <evts>. FD_EV_ERR_* + * doesn't need to also pass FD_EV_SHUT_*, it's implied. ERR and SHUT are + * allowed to be reported regardless of R/W readiness. Returns one of + * FD_UPDT_*. + */ +int fd_update_events(int fd, uint evts) +{ + unsigned long locked; + uint old, new; + uint new_flags, must_stop; + ulong rmask, tmask; + + _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_STUCK); // this thread is still running + + if (unlikely(!fd_grab_tgid(fd, ti->tgid))) { + /* the FD changed to another tgid, we can't safely + * check it anymore. The bits in the masks are not + * ours anymore and we're not allowed to touch them. + * Ours have already been cleared and the FD was + * closed in between so we can safely leave now. + */ + activity[tid].poll_drop_fd++; + return FD_UPDT_CLOSED; + } + + /* Do not take running_mask if not strictly needed (will trigger a + * cosmetic BUG_ON() in fd_insert() anyway if done). + */ + tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask); + if (!(tmask & ti->ltid_bit)) + goto do_update; + + HA_ATOMIC_OR(&fdtab[fd].running_mask, ti->ltid_bit); + + /* From this point, our bit may possibly be in thread_mask, but it may + * still vanish, either because a takeover completed just before taking + * the bit above with the new owner deleting the FD, or because a + * takeover started just before taking the bit. In order to make sure a + * started takeover is complete, we need to verify that all bits of + * running_mask are present in thread_mask, since takeover first takes + * running then atomically replaces thread_mask. Once it's stable, if + * our bit remains there, no further takeover may happen because we + * hold running, but if our bit is not there it means we've lost the + * takeover race and have to decline touching the FD. Regarding the + * risk of deletion, our bit in running_mask prevents fd_delete() from + * finalizing the close, and the caller will leave the FD with a zero + * thread_mask and the FD_MUST_CLOSE flag set. It will then be our + * responsibility to close it. + */ + do { + rmask = _HA_ATOMIC_LOAD(&fdtab[fd].running_mask); + tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask); + rmask &= ~ti->ltid_bit; + } while ((rmask & ~tmask) && (tmask & ti->ltid_bit)); + + /* Now tmask is stable. Do nothing if the FD was taken over under us */ + + if (!(tmask & ti->ltid_bit)) { + /* a takeover has started */ + activity[tid].poll_skip_fd++; + + if (fd_clr_running(fd) == ti->ltid_bit) + goto closed_or_migrated; + + goto do_update; + } + + /* with running we're safe now, we can drop the reference */ + fd_drop_tgid(fd); + + locked = (tmask != ti->ltid_bit); + + /* OK now we are guaranteed that our thread_mask was present and + * that we're allowed to update the FD. + */ + + new_flags = + ((evts & FD_EV_READY_R) ? FD_POLL_IN : 0) | + ((evts & FD_EV_READY_W) ? FD_POLL_OUT : 0) | + ((evts & FD_EV_SHUT_R) ? FD_POLL_HUP : 0) | + ((evts & FD_EV_ERR_RW) ? FD_POLL_ERR : 0); + + /* SHUTW reported while FD was active for writes is an error */ + if ((fdtab[fd].state & FD_EV_ACTIVE_W) && (evts & FD_EV_SHUT_W)) + new_flags |= FD_POLL_ERR; + + /* compute the inactive events reported late that must be stopped */ + must_stop = 0; + if (unlikely(!fd_active(fd))) { + /* both sides stopped */ + must_stop = FD_POLL_IN | FD_POLL_OUT; + } + else if (unlikely(!fd_recv_active(fd) && (evts & (FD_EV_READY_R | FD_EV_SHUT_R | FD_EV_ERR_RW)))) { + /* only send remains */ + must_stop = FD_POLL_IN; + } + else if (unlikely(!fd_send_active(fd) && (evts & (FD_EV_READY_W | FD_EV_SHUT_W | FD_EV_ERR_RW)))) { + /* only recv remains */ + must_stop = FD_POLL_OUT; + } + + if (new_flags & (FD_POLL_IN | FD_POLL_HUP | FD_POLL_ERR)) + new_flags |= FD_EV_READY_R; + + if (new_flags & (FD_POLL_OUT | FD_POLL_ERR)) + new_flags |= FD_EV_READY_W; + + old = fdtab[fd].state; + new = (old & ~FD_POLL_UPDT_MASK) | new_flags; + + if (unlikely(locked)) { + /* Locked FDs (those with more than 2 threads) are atomically updated */ + while (unlikely(new != old && !_HA_ATOMIC_CAS(&fdtab[fd].state, &old, new))) + new = (old & ~FD_POLL_UPDT_MASK) | new_flags; + } else { + if (new != old) + fdtab[fd].state = new; + } + + if (fdtab[fd].iocb && fd_active(fd)) { + fdtab[fd].iocb(fd); + } + + /* + * We entered iocb with running set and with the valid tgid. + * Since then, this is what could have happened: + * - another thread tried to close the FD (e.g. timeout task from + * another one that owns it). We still have running set, but not + * tmask. We must call fd_clr_running() then _fd_delete_orphan() + * if we were the last one. + * + * - the iocb tried to close the FD => bit no more present in running, + * nothing to do. If it managed to close it, the poller's ->clo() + * has already been called. + * + * - after we closed, the FD was reassigned to another thread in + * another group => running not present, tgid differs, nothing to + * do because if it got reassigned it indicates it was already + * closed. + * + * There's no risk of takeover of the valid FD here during this period. + * Also if we still have running, immediately after we release it, the + * events above might instantly happen due to another thread taking + * over. + * + * As such, the only cases where the FD is still relevant are: + * - tgid still set and running still set (most common) + * - tgid still valid but running cleared due to fd_delete(): we may + * still need to stop polling otherwise we may keep it enabled + * while waiting for other threads to close it. + * And given that we may need to program a tentative update in case we + * don't immediately close, it's easier to grab the tgid during the + * whole check. + */ + + if (!fd_grab_tgid(fd, tgid)) + return FD_UPDT_CLOSED; + + tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask); + + /* another thread might have attempted to close this FD in the mean + * time (e.g. timeout task) striking on a previous thread and closing. + * This is detected by us being the last owners of a running_mask bit, + * and the thread_mask being zero. At the moment we release the running + * bit, a takeover may also happen, so in practice we check for our loss + * of the thread_mask bitboth thread_mask and running_mask being 0 after + * we remove ourselves last. There is no risk the FD gets reassigned + * to a different group since it's not released until the real close() + * in _fd_delete_orphan(). + */ + if (fd_clr_running(fd) == ti->ltid_bit && !(tmask & ti->ltid_bit)) + goto closed_or_migrated; + + /* we had to stop this FD and it still must be stopped after the I/O + * cb's changes, so let's program an update for this. + */ + if (must_stop && !(fdtab[fd].update_mask & ti->ltid_bit)) { + if (((must_stop & FD_POLL_IN) && !fd_recv_active(fd)) || + ((must_stop & FD_POLL_OUT) && !fd_send_active(fd))) + if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid)) + fd_updt[fd_nbupdt++] = fd; + } + + fd_drop_tgid(fd); + return FD_UPDT_DONE; + + closed_or_migrated: + /* We only come here once we've last dropped running and the FD is + * not for us as per !(tmask & tid_bit). It may imply we're + * responsible for closing it. Otherwise it's just a migration. + */ + if (HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) { + fd_drop_tgid(fd); + _fd_delete_orphan(fd); + return FD_UPDT_CLOSED; + } + + /* So we were alone, no close bit, at best the FD was migrated, at + * worst it's in the process of being closed by another thread. We must + * be ultra-careful as it can be re-inserted by yet another thread as + * the result of socket() or accept(). Let's just tell the poller the + * FD was lost. If it was closed it was already removed and this will + * only cost an update for nothing. + */ + + do_update: + /* The FD is not closed but we don't want the poller to wake up for + * it anymore. + */ + if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid)) + fd_updt[fd_nbupdt++] = fd; + + fd_drop_tgid(fd); + return FD_UPDT_MIGRATED; +} + +/* This is used by pollers at boot time to re-register desired events for + * all FDs after new pollers have been created. It doesn't do much, it checks + * that their thread group matches the one in argument, and that the thread + * mask matches at least one of the bits in the mask, and if so, marks the FD + * as updated. + */ +void fd_reregister_all(int tgrp, ulong mask) +{ + int fd; + + for (fd = 0; fd < global.maxsock; fd++) { + if (!fdtab[fd].owner) + continue; + + /* make sure we don't register other tgroups' FDs. We just + * avoid needlessly taking the lock if not needed. + */ + if (!(_HA_ATOMIC_LOAD(&fdtab[fd].thread_mask) & mask) || + !fd_grab_tgid(fd, tgrp)) + continue; // was not for us anyway + + if (_HA_ATOMIC_LOAD(&fdtab[fd].thread_mask) & mask) + updt_fd_polling(fd); + fd_drop_tgid(fd); + } +} + +/* Tries to send <npfx> parts from <prefix> followed by <nmsg> parts from <msg> + * optionally followed by a newline if <nl> is non-null, to file descriptor + * <fd>. The message is sent atomically using writev(). It may be truncated to + * <maxlen> bytes if <maxlen> is non-null. There is no distinction between the + * two lists, it's just a convenience to help the caller prepend some prefixes + * when necessary. It takes the fd's lock to make sure no other thread will + * write to the same fd in parallel. Returns the number of bytes sent, or <=0 + * on failure. A limit to 31 total non-empty segments is enforced. The caller + * is responsible for taking care of making the fd non-blocking. + */ +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) +{ + struct iovec iovec[32]; + size_t sent = 0; + int vec = 0; + int attempts = 0; + + if (!maxlen) + maxlen = ~0; + + /* keep one char for a possible trailing '\n' in any case */ + maxlen--; + + /* make an iovec from the concatenation of all parts of the original + * message. Skip empty fields and truncate the whole message to maxlen, + * leaving one spare iovec for the '\n'. + */ + while (vec < (sizeof(iovec) / sizeof(iovec[0]) - 1)) { + if (!npfx) { + pfx = msg; + npfx = nmsg; + nmsg = 0; + if (!npfx) + break; + } + + iovec[vec].iov_base = pfx->ptr; + iovec[vec].iov_len = MIN(maxlen, pfx->len); + maxlen -= iovec[vec].iov_len; + if (iovec[vec].iov_len) + vec++; + pfx++; npfx--; + }; + + if (nl) { + iovec[vec].iov_base = "\n"; + iovec[vec].iov_len = 1; + vec++; + } + + /* make sure we never interleave writes and we never block. This means + * we prefer to fail on collision than to block. But we don't want to + * lose too many logs so we just perform a few lock attempts then give + * up. + */ + + while (HA_ATOMIC_BTS(&fdtab[fd].state, FD_EXCL_SYSCALL_BIT)) { + if (++attempts >= 200) { + /* so that the caller knows the message couldn't be delivered */ + sent = -1; + errno = EAGAIN; + goto leave; + } + ha_thread_relax(); + } + + if (unlikely(!(fdtab[fd].state & FD_INITIALIZED))) { + HA_ATOMIC_OR(&fdtab[fd].state, FD_INITIALIZED); + if (!isatty(fd)) + fd_set_nonblock(fd); + } + sent = writev(fd, iovec, vec); + HA_ATOMIC_BTR(&fdtab[fd].state, FD_EXCL_SYSCALL_BIT); + + leave: + /* sent > 0 if the message was delivered */ + return sent; +} + +#if defined(USE_CLOSEFROM) +void my_closefrom(int start) +{ + closefrom(start); +} + +#elif defined(USE_POLL) +/* This is a portable implementation of closefrom(). It closes all open file + * descriptors starting at <start> and above. It relies on the fact that poll() + * will return POLLNVAL for each invalid (hence close) file descriptor passed + * in argument in order to skip them. It acts with batches of FDs and will + * typically perform one poll() call per 1024 FDs so the overhead is low in + * case all FDs have to be closed. + */ +void my_closefrom(int start) +{ + struct pollfd poll_events[1024]; + struct rlimit limit; + int nbfds, fd, ret, idx; + int step, next; + + if (getrlimit(RLIMIT_NOFILE, &limit) == 0) + step = nbfds = limit.rlim_cur; + else + step = nbfds = 0; + + if (nbfds <= 0) { + /* set safe limit */ + nbfds = 1024; + step = 256; + } + + if (step > sizeof(poll_events) / sizeof(poll_events[0])) + step = sizeof(poll_events) / sizeof(poll_events[0]); + + while (start < nbfds) { + next = (start / step + 1) * step; + + for (fd = start; fd < next && fd < nbfds; fd++) { + poll_events[fd - start].fd = fd; + poll_events[fd - start].events = 0; + } + + do { + ret = poll(poll_events, fd - start, 0); + if (ret >= 0) + break; + } while (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR || errno == ENOMEM); + + if (ret) + ret = fd - start; + + for (idx = 0; idx < ret; idx++) { + if (poll_events[idx].revents & POLLNVAL) + continue; /* already closed */ + + fd = poll_events[idx].fd; + close(fd); + } + start = next; + } +} + +#else // defined(USE_POLL) + +/* This is a portable implementation of closefrom(). It closes all open file + * descriptors starting at <start> and above. This is a naive version for use + * when the operating system provides no alternative. + */ +void my_closefrom(int start) +{ + struct rlimit limit; + int nbfds; + + if (getrlimit(RLIMIT_NOFILE, &limit) == 0) + nbfds = limit.rlim_cur; + else + nbfds = 0; + + if (nbfds <= 0) + nbfds = 1024; /* safe limit */ + + while (start < nbfds) + close(start++); +} +#endif // defined(USE_POLL) + +/* Sets the RLIMIT_NOFILE setting to <new_limit> and returns the previous one + * in <old_limit> if the pointer is not NULL, even if set_rlimit() fails. The + * two pointers may point to the same variable as the copy happens after + * setting the new value. The value is only changed if at least one of the new + * limits is strictly higher than the current one, otherwise returns 0 without + * changing anything. The getrlimit() or setrlimit() syscall return value is + * returned and errno is preserved. + */ +int raise_rlim_nofile(struct rlimit *old_limit, struct rlimit *new_limit) +{ + struct rlimit limit = { }; + int ret = 0; + + ret = getrlimit(RLIMIT_NOFILE, &limit); + + if (ret == 0 && + (limit.rlim_max < new_limit->rlim_max || + limit.rlim_cur < new_limit->rlim_cur)) { + ret = setrlimit(RLIMIT_NOFILE, new_limit); + } + + if (old_limit) + *old_limit = limit; + + return ret; +} + +/* Computes the bounded poll() timeout based on the next expiration timer <next> + * by bounding it to MAX_DELAY_MS. <next> may equal TICK_ETERNITY. The pollers + * just needs to call this function right before polling to get their timeout + * value. Timeouts that are already expired (possibly due to a pending event) + * are accounted for in activity.poll_exp. + */ +int compute_poll_timeout(int next) +{ + int wait_time; + + if (!tick_isset(next)) + wait_time = MAX_DELAY_MS; + else if (tick_is_expired(next, now_ms)) { + activity[tid].poll_exp++; + wait_time = 0; + } + else { + wait_time = TICKS_TO_MS(tick_remain(now_ms, next)) + 1; + if (wait_time > MAX_DELAY_MS) + wait_time = MAX_DELAY_MS; + } + return wait_time; +} + +/* Handle the return of the poller, which consists in calculating the idle + * time, saving a few clocks, marking the thread harmful again etc. All that + * is some boring stuff that all pollers have to do anyway. + */ +void fd_leaving_poll(int wait_time, int status) +{ + clock_leaving_poll(wait_time, status); + + thread_harmless_end(); + thread_idle_end(); + + _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_SLEEPING); +} + +/* disable the specified poller */ +void disable_poller(const char *poller_name) +{ + int p; + + for (p = 0; p < nbpollers; p++) + if (strcmp(pollers[p].name, poller_name) == 0) + pollers[p].pref = 0; +} + +void poller_pipe_io_handler(int fd) +{ + char buf[1024]; + /* Flush the pipe */ + while (read(fd, buf, sizeof(buf)) > 0); + fd_cant_recv(fd); +} + +/* allocate the per-thread fd_updt thus needs to be called early after + * thread creation. + */ +static int alloc_pollers_per_thread() +{ + fd_updt = calloc(global.maxsock, sizeof(*fd_updt)); + return fd_updt != NULL; +} + +/* Initialize the pollers per thread.*/ +static int init_pollers_per_thread() +{ + int mypipe[2]; + + if (pipe(mypipe) < 0) + return 0; + + poller_rd_pipe = mypipe[0]; + poller_wr_pipe[tid] = mypipe[1]; + fd_set_nonblock(poller_rd_pipe); + fd_insert(poller_rd_pipe, poller_pipe_io_handler, poller_pipe_io_handler, tgid, ti->ltid_bit); + fd_insert(poller_wr_pipe[tid], poller_pipe_io_handler, poller_pipe_io_handler, tgid, ti->ltid_bit); + fd_want_recv(poller_rd_pipe); + fd_stop_both(poller_wr_pipe[tid]); + return 1; +} + +/* Deinitialize the pollers per thread */ +static void deinit_pollers_per_thread() +{ + /* rd and wr are init at the same place, but only rd is init to -1, so + we rely to rd to close. */ + if (poller_rd_pipe > -1) { + fd_delete(poller_rd_pipe); + poller_rd_pipe = -1; + fd_delete(poller_wr_pipe[tid]); + poller_wr_pipe[tid] = -1; + } +} + +/* Release the pollers per thread, to be called late */ +static void free_pollers_per_thread() +{ + fd_nbupdt = 0; + ha_free(&fd_updt); +} + +/* + * Initialize the pollers till the best one is found. + * If none works, returns 0, otherwise 1. + */ +int init_pollers() +{ + int p; + struct poller *bp; + + if ((fdtab_addr = calloc(global.maxsock, sizeof(*fdtab) + 64)) == NULL) { + ha_alert("Not enough memory to allocate %d entries for fdtab!\n", global.maxsock); + goto fail_tab; + } + + /* always provide an aligned fdtab */ + fdtab = (struct fdtab*)((((size_t)fdtab_addr) + 63) & -(size_t)64); + + if ((polled_mask = calloc(global.maxsock, sizeof(*polled_mask))) == NULL) { + ha_alert("Not enough memory to allocate %d entries for polled_mask!\n", global.maxsock); + goto fail_polledmask; + } + + if ((fdinfo = calloc(global.maxsock, sizeof(*fdinfo))) == NULL) { + ha_alert("Not enough memory to allocate %d entries for fdinfo!\n", global.maxsock); + goto fail_info; + } + + for (p = 0; p < MAX_TGROUPS; p++) + update_list[p].first = update_list[p].last = -1; + + for (p = 0; p < global.maxsock; p++) { + /* Mark the fd as out of the fd cache */ + fdtab[p].update.next = -3; + } + + do { + bp = NULL; + for (p = 0; p < nbpollers; p++) + if (!bp || (pollers[p].pref > bp->pref)) + bp = &pollers[p]; + + if (!bp || bp->pref == 0) + break; + + if (bp->init(bp)) { + memcpy(&cur_poller, bp, sizeof(*bp)); + return 1; + } + } while (!bp || bp->pref == 0); + + free(fdinfo); + fail_info: + free(polled_mask); + fail_polledmask: + free(fdtab_addr); + fail_tab: + return 0; +} + +/* + * Deinitialize the pollers. + */ +void deinit_pollers() { + + struct poller *bp; + int p; + + for (p = 0; p < nbpollers; p++) { + bp = &pollers[p]; + + if (bp && bp->pref) + bp->term(bp); + } + + ha_free(&fdinfo); + ha_free(&fdtab_addr); + ha_free(&polled_mask); +} + +/* + * Lists the known pollers on <out>. + * Should be performed only before initialization. + */ +int list_pollers(FILE *out) +{ + int p; + int last, next; + int usable; + struct poller *bp; + + fprintf(out, "Available polling systems :\n"); + + usable = 0; + bp = NULL; + last = next = -1; + while (1) { + for (p = 0; p < nbpollers; p++) { + if ((next < 0 || pollers[p].pref > next) + && (last < 0 || pollers[p].pref < last)) { + next = pollers[p].pref; + if (!bp || (pollers[p].pref > bp->pref)) + bp = &pollers[p]; + } + } + + if (next == -1) + break; + + for (p = 0; p < nbpollers; p++) { + if (pollers[p].pref == next) { + fprintf(out, " %10s : ", pollers[p].name); + if (pollers[p].pref == 0) + fprintf(out, "disabled, "); + else + fprintf(out, "pref=%3d, ", pollers[p].pref); + if (pollers[p].test(&pollers[p])) { + fprintf(out, " test result OK"); + if (next > 0) + usable++; + } else { + fprintf(out, " test result FAILED"); + if (bp == &pollers[p]) + bp = NULL; + } + fprintf(out, "\n"); + } + } + last = next; + next = -1; + }; + fprintf(out, "Total: %d (%d usable), will use %s.\n", nbpollers, usable, bp ? bp->name : "none"); + return 0; +} + +/* + * Some pollers may lose their connection after a fork(). It may be necessary + * to create initialize part of them again. Returns 0 in case of failure, + * otherwise 1. The fork() function may be NULL if unused. In case of error, + * the the current poller is destroyed and the caller is responsible for trying + * another one by calling init_pollers() again. + */ +int fork_poller() +{ + int fd; + for (fd = 0; fd < global.maxsock; fd++) { + if (fdtab[fd].owner) { + HA_ATOMIC_OR(&fdtab[fd].state, FD_CLONED); + } + } + + if (cur_poller.fork) { + if (cur_poller.fork(&cur_poller)) + return 1; + cur_poller.term(&cur_poller); + return 0; + } + return 1; +} + +/* config parser for global "tune.fd.edge-triggered", accepts "on" or "off" */ +static int cfg_parse_tune_fd_edge_triggered(char **args, int section_type, struct proxy *curpx, + const struct proxy *defpx, const char *file, int line, + char **err) +{ + if (too_many_args(1, args, err, NULL)) + return -1; + + if (strcmp(args[1], "on") == 0) + global.tune.options |= GTUNE_FD_ET; + else if (strcmp(args[1], "off") == 0) + global.tune.options &= ~GTUNE_FD_ET; + else { + memprintf(err, "'%s' expects either 'on' or 'off' but got '%s'.", args[0], args[1]); + return -1; + } + return 0; +} + +/* config keyword parsers */ +static struct cfg_kw_list cfg_kws = {ILH, { + { CFG_GLOBAL, "tune.fd.edge-triggered", cfg_parse_tune_fd_edge_triggered, KWF_EXPERIMENTAL }, + { 0, NULL, NULL } +}}; + +INITCALL1(STG_REGISTER, cfg_register_keywords, &cfg_kws); + +REGISTER_PER_THREAD_ALLOC(alloc_pollers_per_thread); +REGISTER_PER_THREAD_INIT(init_pollers_per_thread); +REGISTER_PER_THREAD_DEINIT(deinit_pollers_per_thread); +REGISTER_PER_THREAD_FREE(free_pollers_per_thread); + +/* + * Local variables: + * c-indent-level: 8 + * c-basic-offset: 8 + * End: + */ |