| /* |
| * include/proto/fd.h |
| * File descriptors states. |
| * |
| * Copyright (C) 2000-2014 Willy Tarreau - w@1wt.eu |
| * |
| * This library is free software; you can redistribute it and/or |
| * modify it under the terms of the GNU Lesser General Public |
| * License as published by the Free Software Foundation, version 2.1 |
| * exclusively. |
| * |
| * This library is distributed in the hope that it will be useful, |
| * but WITHOUT ANY WARRANTY; without even the implied warranty of |
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| * Lesser General Public License for more details. |
| * |
| * You should have received a copy of the GNU Lesser General Public |
| * License along with this library; if not, write to the Free Software |
| * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA |
| */ |
| |
| #ifndef _PROTO_FD_H |
| #define _PROTO_FD_H |
| |
| #include <stdio.h> |
| #include <sys/time.h> |
| #include <sys/types.h> |
| #include <unistd.h> |
| |
| #include <common/config.h> |
| #include <common/ticks.h> |
| #include <common/time.h> |
| #include <types/fd.h> |
| #include <proto/activity.h> |
| |
| /* public variables */ |
| |
| extern volatile struct fdlist update_list; |
| |
| |
| extern struct polled_mask { |
| unsigned long poll_recv; |
| unsigned long poll_send; |
| } *polled_mask; |
| |
| extern THREAD_LOCAL int *fd_updt; // FD updates list |
| extern THREAD_LOCAL int fd_nbupdt; // number of updates in the list |
| |
| extern int poller_wr_pipe[MAX_THREADS]; |
| |
| extern volatile int ha_used_fds; // Number of FDs we're currently using |
| |
| /* Deletes an FD from the fdsets. |
| * The file descriptor is also closed. |
| */ |
| void fd_delete(int fd); |
| |
| /* Deletes an FD from the fdsets. |
| * The file descriptor is kept open. |
| */ |
| void fd_remove(int fd); |
| |
| 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); |
| |
| /* close all FDs starting from <start> */ |
| void my_closefrom(int start); |
| |
| /* disable the specified poller */ |
| void disable_poller(const char *poller_name); |
| |
| void poller_pipe_io_handler(int fd); |
| |
| /* |
| * Initialize the pollers till the best one is found. |
| * If none works, returns 0, otherwise 1. |
| * The pollers register themselves just before main() is called. |
| */ |
| int init_pollers(); |
| |
| /* |
| * Deinitialize the pollers. |
| */ |
| void deinit_pollers(); |
| |
| /* |
| * 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(); |
| |
| /* |
| * Lists the known pollers on <out>. |
| * Should be performed only before initialization. |
| */ |
| int list_pollers(FILE *out); |
| |
| /* |
| * Runs the polling loop |
| */ |
| void run_poller(); |
| |
| void fd_add_to_fd_list(volatile struct fdlist *list, int fd, int off); |
| void fd_rm_from_fd_list(volatile struct fdlist *list, int fd, int off); |
| void updt_fd_polling(const int fd); |
| |
| /* Called from the poller to acknoledge we read an entry from the global |
| * update list, to remove our bit from the update_mask, and remove it from |
| * the list if we were the last one. |
| */ |
| static inline void done_update_polling(int fd) |
| { |
| unsigned long update_mask; |
| |
| update_mask = _HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tid_bit); |
| while ((update_mask & all_threads_mask)== 0) { |
| /* If we were the last one that had to update that entry, remove it from the list */ |
| fd_rm_from_fd_list(&update_list, fd, offsetof(struct fdtab, update)); |
| update_mask = (volatile unsigned long)fdtab[fd].update_mask; |
| if ((update_mask & all_threads_mask) != 0) { |
| /* Maybe it's been re-updated in the meanwhile, and we |
| * wrongly removed it from the list, if so, re-add it |
| */ |
| fd_add_to_fd_list(&update_list, fd, offsetof(struct fdtab, update)); |
| update_mask = (volatile unsigned long)(fdtab[fd].update_mask); |
| /* And then check again, just in case after all it |
| * should be removed, even if it's very unlikely, given |
| * the current thread wouldn't have been able to take |
| * care of it yet */ |
| } else |
| break; |
| |
| } |
| } |
| |
| /* |
| * returns true if the FD is active for recv |
| */ |
| static inline int fd_recv_active(const int fd) |
| { |
| return (unsigned)fdtab[fd].state & FD_EV_ACTIVE_R; |
| } |
| |
| /* |
| * returns true if the FD is ready for recv |
| */ |
| static inline int fd_recv_ready(const int fd) |
| { |
| return (unsigned)fdtab[fd].state & FD_EV_READY_R; |
| } |
| |
| /* |
| * returns true if the FD is active for send |
| */ |
| static inline int fd_send_active(const int fd) |
| { |
| return (unsigned)fdtab[fd].state & FD_EV_ACTIVE_W; |
| } |
| |
| /* |
| * returns true if the FD is ready for send |
| */ |
| static inline int fd_send_ready(const int fd) |
| { |
| return (unsigned)fdtab[fd].state & FD_EV_READY_W; |
| } |
| |
| /* |
| * returns true if the FD is active for recv or send |
| */ |
| static inline int fd_active(const int fd) |
| { |
| return (unsigned)fdtab[fd].state & FD_EV_ACTIVE_RW; |
| } |
| |
| /* Disable processing recv events on fd <fd> */ |
| static inline void fd_stop_recv(int fd) |
| { |
| if (!(fdtab[fd].state & FD_EV_ACTIVE_R) || |
| !HA_ATOMIC_BTR(&fdtab[fd].state, FD_EV_ACTIVE_R_BIT)) |
| return; |
| } |
| |
| /* Disable processing send events on fd <fd> */ |
| static inline void fd_stop_send(int fd) |
| { |
| if (!(fdtab[fd].state & FD_EV_ACTIVE_W) || |
| !HA_ATOMIC_BTR(&fdtab[fd].state, FD_EV_ACTIVE_W_BIT)) |
| return; |
| } |
| |
| /* Disable processing of events on fd <fd> for both directions. */ |
| static inline void fd_stop_both(int fd) |
| { |
| unsigned char old, new; |
| |
| old = fdtab[fd].state; |
| do { |
| if (!(old & FD_EV_ACTIVE_RW)) |
| return; |
| new = old & ~FD_EV_ACTIVE_RW; |
| } while (unlikely(!_HA_ATOMIC_CAS(&fdtab[fd].state, &old, new))); |
| } |
| |
| /* Report that FD <fd> cannot receive anymore without polling (EAGAIN detected). */ |
| static inline void fd_cant_recv(const int fd) |
| { |
| /* marking ready never changes polled status */ |
| if (!(fdtab[fd].state & FD_EV_READY_R) || |
| !HA_ATOMIC_BTR(&fdtab[fd].state, FD_EV_READY_R_BIT)) |
| return; |
| } |
| |
| /* Report that FD <fd> may receive again without polling. */ |
| static inline void fd_may_recv(const int fd) |
| { |
| /* marking ready never changes polled status */ |
| if ((fdtab[fd].state & FD_EV_READY_R) || |
| HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_READY_R_BIT)) |
| return; |
| } |
| |
| /* Report that FD <fd> may receive again without polling but only if its not |
| * active yet. This is in order to speculatively try to enable I/Os when it's |
| * highly likely that these will succeed, but without interfering with polling. |
| */ |
| static inline void fd_cond_recv(const int fd) |
| { |
| if ((fdtab[fd].state & (FD_EV_ACTIVE_R|FD_EV_READY_R)) == 0) |
| HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_READY_R_BIT); |
| } |
| |
| /* Report that FD <fd> may send again without polling but only if its not |
| * active yet. This is in order to speculatively try to enable I/Os when it's |
| * highly likely that these will succeed, but without interfering with polling. |
| */ |
| static inline void fd_cond_send(const int fd) |
| { |
| if ((fdtab[fd].state & (FD_EV_ACTIVE_W|FD_EV_READY_W)) == 0) |
| HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_READY_W_BIT); |
| } |
| |
| /* Report that FD <fd> may receive and send without polling. Used at FD |
| * initialization. |
| */ |
| static inline void fd_may_both(const int fd) |
| { |
| HA_ATOMIC_OR(&fdtab[fd].state, FD_EV_READY_RW); |
| } |
| |
| /* Disable readiness when active. This is useful to interrupt reading when it |
| * is suspected that the end of data might have been reached (eg: short read). |
| * This can only be done using level-triggered pollers, so if any edge-triggered |
| * is ever implemented, a test will have to be added here. |
| */ |
| static inline void fd_done_recv(const int fd) |
| { |
| /* removing ready never changes polled status */ |
| if ((fdtab[fd].state & (FD_EV_ACTIVE_R|FD_EV_READY_R)) != (FD_EV_ACTIVE_R|FD_EV_READY_R) || |
| !HA_ATOMIC_BTR(&fdtab[fd].state, FD_EV_READY_R_BIT)) |
| return; |
| } |
| |
| /* Report that FD <fd> cannot send anymore without polling (EAGAIN detected). */ |
| static inline void fd_cant_send(const int fd) |
| { |
| /* removing ready never changes polled status */ |
| if (!(fdtab[fd].state & FD_EV_READY_W) || |
| !HA_ATOMIC_BTR(&fdtab[fd].state, FD_EV_READY_W_BIT)) |
| return; |
| } |
| |
| /* Report that FD <fd> may send again without polling (EAGAIN not detected). */ |
| static inline void fd_may_send(const int fd) |
| { |
| /* marking ready never changes polled status */ |
| if ((fdtab[fd].state & FD_EV_READY_W) || |
| HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_READY_W_BIT)) |
| return; |
| } |
| |
| /* Prepare FD <fd> to try to receive */ |
| static inline void fd_want_recv(int fd) |
| { |
| if ((fdtab[fd].state & FD_EV_ACTIVE_R) || |
| HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_ACTIVE_R_BIT)) |
| return; |
| updt_fd_polling(fd); |
| } |
| |
| /* Prepare FD <fd> to try to send */ |
| static inline void fd_want_send(int fd) |
| { |
| if ((fdtab[fd].state & FD_EV_ACTIVE_W) || |
| HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_ACTIVE_W_BIT)) |
| return; |
| updt_fd_polling(fd); |
| } |
| |
| /* 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. |
| */ |
| static inline void fd_update_events(int fd, unsigned char evts) |
| { |
| unsigned long locked = atleast2(fdtab[fd].thread_mask); |
| unsigned char old, new; |
| int new_flags, must_stop; |
| |
| 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_R) ? FD_POLL_ERR : 0) | |
| ((evts & FD_EV_ERR_W) ? FD_POLL_ERR : 0); |
| |
| /* SHUTW reported while FD was active for writes is an error */ |
| if ((fdtab[fd].ev & 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_R)))) { |
| /* 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_W)))) { |
| /* only recv remains */ |
| must_stop = FD_POLL_OUT; |
| } |
| |
| old = fdtab[fd].ev; |
| new = (old & FD_POLL_STICKY) | 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].ev, &old, new))) |
| new = (old & FD_POLL_STICKY) | new_flags; |
| } else { |
| if (new != old) |
| fdtab[fd].ev = new; |
| } |
| |
| if (fdtab[fd].ev & (FD_POLL_IN | FD_POLL_HUP | FD_POLL_ERR)) |
| fd_may_recv(fd); |
| |
| if (fdtab[fd].ev & (FD_POLL_OUT | FD_POLL_ERR)) |
| fd_may_send(fd); |
| |
| if (fdtab[fd].iocb && fd_active(fd)) |
| fdtab[fd].iocb(fd); |
| |
| /* 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 & tid_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, tid)) |
| fd_updt[fd_nbupdt++] = fd; |
| } |
| |
| ti->flags &= ~TI_FL_STUCK; // this thread is still running |
| } |
| |
| /* Prepares <fd> for being polled */ |
| static inline void fd_insert(int fd, void *owner, void (*iocb)(int fd), unsigned long thread_mask) |
| { |
| unsigned long locked = atleast2(thread_mask); |
| |
| if (locked) |
| HA_SPIN_LOCK(FD_LOCK, &fdtab[fd].lock); |
| fdtab[fd].owner = owner; |
| fdtab[fd].iocb = iocb; |
| fdtab[fd].ev = 0; |
| fdtab[fd].linger_risk = 0; |
| fdtab[fd].cloned = 0; |
| fdtab[fd].thread_mask = thread_mask; |
| /* note: do not reset polled_mask here as it indicates which poller |
| * still knows this FD from a possible previous round. |
| */ |
| if (locked) |
| HA_SPIN_UNLOCK(FD_LOCK, &fdtab[fd].lock); |
| /* the two directions are ready until proven otherwise */ |
| fd_may_both(fd); |
| _HA_ATOMIC_ADD(&ha_used_fds, 1); |
| } |
| |
| /* 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. |
| */ |
| static inline 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; |
| } |
| |
| /* These are replacements for FD_SET, FD_CLR, FD_ISSET, working on uints */ |
| static inline void hap_fd_set(int fd, unsigned int *evts) |
| { |
| _HA_ATOMIC_OR(&evts[fd / (8*sizeof(*evts))], 1U << (fd & (8*sizeof(*evts) - 1))); |
| } |
| |
| static inline void hap_fd_clr(int fd, unsigned int *evts) |
| { |
| _HA_ATOMIC_AND(&evts[fd / (8*sizeof(*evts))], ~(1U << (fd & (8*sizeof(*evts) - 1)))); |
| } |
| |
| static inline unsigned int hap_fd_isset(int fd, unsigned int *evts) |
| { |
| return evts[fd / (8*sizeof(*evts))] & (1U << (fd & (8*sizeof(*evts) - 1))); |
| } |
| |
| static inline void wake_thread(int tid) |
| { |
| char c = 'c'; |
| |
| shut_your_big_mouth_gcc(write(poller_wr_pipe[tid], &c, 1)); |
| } |
| |
| |
| #endif /* _PROTO_FD_H */ |
| |
| /* |
| * Local variables: |
| * c-indent-level: 8 |
| * c-basic-offset: 8 |
| * End: |
| */ |