/src/haproxy/include/haproxy/fd.h

Source
/*
 * include/haproxy/fd.h
 * File descriptors states - exported variables and functions
 *
 * Copyright (C) 2000-2020 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 _HAPROXY_FD_H
#define _HAPROXY_FD_H

#include <sys/time.h>
#include <sys/types.h>
#include <stdio.h>
#include <unistd.h>
#include <import/ist.h>
#include <haproxy/api.h>
#include <haproxy/atomic.h>
#include <haproxy/fd-t.h>
#include <haproxy/global.h>
#include <haproxy/thread.h>

/* public variables */

extern struct poller cur_poller; /* the current poller */
extern int nbpollers;
extern struct poller pollers[MAX_POLLERS];   /* all registered pollers */
extern THREAD_LOCAL struct fdtab *fdtab;     /* array of all the file descriptors */
extern int totalconn;                   /* total # of terminated sessions */
extern int actconn;                     /* # of active sessions */

extern volatile struct fdlist update_list[MAX_TGROUPS];
extern THREAD_LOCAL struct polled_mask *polled_mask; /* Array for the polled_mask of each fd */

extern THREAD_LOCAL int *fd_updt;  // FD updates list
extern THREAD_LOCAL int fd_nbupdt; // number of updates in the list
extern THREAD_LOCAL int fd_highest;// highest FD known by the current thread

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);
void _fd_delete_orphan(int 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.
 */
int fd_set_nonblock(int fd);

/* makes the fd close-on-exec; returns -1 on failure. */
int fd_set_cloexec(int fd);

/* Migrate a FD to a new thread <new_tid>. */
void fd_migrate_on(int fd, uint new_tid);

/*
 * Take over a FD belonging to another thread.
 * Returns 0 on success, and -1 on failure.
 */
int fd_takeover(int fd, void *expected_owner);

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);

int compute_poll_timeout(int next);
void fd_leaving_poll(int wait_time, int status);

/* 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(void);

int fd_precreate_poller_pipes(void);

/*
 * Deinitialize the pollers.
 */
void deinit_pollers(void);

/*
 * 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(void);

/*
 * 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);
void fd_rm_from_fd_list(volatile struct fdlist *list, int fd);
void updt_fd_polling(const int fd);
int fd_update_events(int fd, uint evts);
void fd_reregister_all(int tgrp, ulong mask);

/* Called from the poller to acknowledge 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_FETCH(&fdtab[fd].update_mask, ~ti->ltid_bit);
  while ((update_mask & _HA_ATOMIC_LOAD(&tg->threads_enabled)) == 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[tgid - 1], fd);
    update_mask = _HA_ATOMIC_LOAD(&fdtab[fd].update_mask);
    if ((update_mask & _HA_ATOMIC_LOAD(&tg->threads_enabled)) != 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[tgid - 1], fd);
      update_mask = _HA_ATOMIC_LOAD(&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)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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)
{
  uint old, new;

  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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 */
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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 */
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  HA_ATOMIC_OR(&fdtab[fd].state, FD_EV_READY_RW);
}

/* 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 */
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  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 receive, and only create update if fd_updt exists
 * (essentially for receivers during early boot).
 */
static inline void fd_want_recv_safe(int fd)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  if ((fdtab[fd].state & FD_EV_ACTIVE_R) ||
      HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_ACTIVE_R_BIT))
    return;
  if (fd_updt)
    updt_fd_polling(fd);
}

/* Prepare FD <fd> to try to send */
static inline void fd_want_send(int fd)
{
  BUG_ON_HOT(fdtab[fd].owner == NULL);
  if ((fdtab[fd].state & FD_EV_ACTIVE_W) ||
      HA_ATOMIC_BTS(&fdtab[fd].state, FD_EV_ACTIVE_W_BIT))
    return;
  updt_fd_polling(fd);
}

/* returns the tgid from an fd (masks the refcount) */
static forceinline int fd_tgid(int fd)
{
  return _HA_ATOMIC_LOAD(&fdtab[fd].refc_tgid) & 0xFFFF;
}

/* Release a tgid previously taken by fd_grab_tgid() */
static forceinline void fd_drop_tgid(int fd)
{
  HA_ATOMIC_SUB(&fdtab[fd].refc_tgid, 0x10000);
}

/* Unlock a tgid currently locked by fd_lock_tgid(). This will effectively
 * allow threads from the FD's tgid to check the masks and manipulate the FD.
 */
static forceinline void fd_unlock_tgid(int fd)
{
  HA_ATOMIC_AND(&fdtab[fd].refc_tgid, 0xffff7fffU);
}

/* Switch the FD's TGID to the new value with a refcount of 1 and the lock bit
 * set. It doesn't care about the current TGID, except that it will wait for
 * the FD not to be already switching and having its refcount cleared. After
 * the function returns, the caller is free to manipulate the masks, and it
 * must call fd_unlock_tgid() to drop the lock, allowing threads from the
 * designated group to use the FD. Finally a call to fd_drop_tgid() will be
 * needed to drop the reference.
 */
static inline void fd_lock_tgid(int fd, uint desired_tgid)
{
  uint old;

  BUG_ON(!desired_tgid);

  old = tgid;  // assume we start from the caller's tgid
  desired_tgid |= 0x18000; // refcount=1, lock bit=1.

  while (1) {
    old &= 0x7fff; // expect no lock and refcount==0
    if (_HA_ATOMIC_CAS(&fdtab[fd].refc_tgid, &old, desired_tgid))
      break;
    __ha_cpu_relax();
  }
}

/*
 * Try to lock the tgid, keeping the current tgid value.
 * Returns 1 on success, or 0 on failure.
 */
static inline int fd_lock_tgid_cur(int fd)
{
  uint old = _HA_ATOMIC_LOAD(&fdtab[fd].refc_tgid) & 0x7fff;

  if (_HA_ATOMIC_CAS(&fdtab[fd].refc_tgid, &old, (old | 0x8000) + 0x10000))
    return 1;
  return 0;
}


/* Grab a reference to the FD's TGID, and return the tgid. Note that a TGID of
 * zero indicates the FD was closed, thus also fails (i.e. no need to drop it).
 * On non-zero (success), the caller must release it using fd_drop_tgid().
 */
static inline uint fd_take_tgid(int fd)
{
  uint old;

  old = _HA_ATOMIC_FETCH_ADD(&fdtab[fd].refc_tgid, 0x10000) & 0xffff;
  while (old & 0x8000) {
                old = _HA_ATOMIC_LOAD(&fdtab[fd].refc_tgid) & 0xffff;
                __ha_cpu_relax();
  }
  if (likely(old))
    return old;
  HA_ATOMIC_SUB(&fdtab[fd].refc_tgid, 0x10000);
  return 0;
}

/* Reset a tgid without affecting the refcount */
static forceinline void fd_reset_tgid(int fd)
{
  HA_ATOMIC_AND(&fdtab[fd].refc_tgid, 0xffff0000U);
}

/* Try to grab a reference to the FD's TGID, but only if it matches the
 * requested one (i.e. it succeeds with TGID refcnt held, or fails). Note that
 * a TGID of zero indicates the FD was closed, thus also fails. It returns
 * non-zero on success, in which case the caller must then release it using
 * fd_drop_tgid(), or zero on failure. The function is optimized for use
 * when it's likely that the tgid matches the desired one as it's by far
 * the most common.
 */
static inline uint fd_grab_tgid(int fd, uint desired_tgid)
{
  uint old;

  old = _HA_ATOMIC_FETCH_ADD(&fdtab[fd].refc_tgid, 0x10000) & 0xffff;
  /* If the tgid is locked, wait until it no longer is */
  while (old & 0x8000) {
    old = _HA_ATOMIC_LOAD(&fdtab[fd].refc_tgid) & 0xffff;
    __ha_cpu_relax();
  }
  if (likely(old == desired_tgid))
    return 1;
  HA_ATOMIC_SUB(&fdtab[fd].refc_tgid, 0x10000);
  return 0;
}

/* Set the FD's TGID to the new value with a refcount of 1, waiting for the
 * current refcount to become 0, to cover the rare possibly that a late
 * competing thread would be touching the tgid or the running mask in parallel.
 * The caller must call fd_drop_tgid() once done.
 */
static inline void fd_claim_tgid(int fd, uint desired_tgid)
{
  uint old;

  BUG_ON(!desired_tgid);

  desired_tgid += 0x10000; // refcount=1
  old = 0;                 // assume unused (most likely)
  while (1) {
    if (_HA_ATOMIC_CAS(&fdtab[fd].refc_tgid, &old, desired_tgid))
      break;
    __ha_cpu_relax();
    old &= 0x7fff;   // keep only the tgid and drop the lock
  }
}

/*
 * Update the FD's TGID.
 * This should be called with the lock held, and will drop the lock once
 * the TGID is updated.
 * The reference counter is however preserved.
 */
static inline void fd_update_tgid(int fd, uint desired_tgid)
{
  unsigned int orig_tgid = fdtab[fd].refc_tgid;
  unsigned int new_tgid;
  /* Remove the lock, and switch to the new tgid */
  do {
    new_tgid = (orig_tgid & 0xffff0000) | desired_tgid;
  } while (!_HA_ATOMIC_CAS(&fdtab[fd].refc_tgid, &orig_tgid, new_tgid) && __ha_cpu_relax());
}

/* atomically read the running mask if the tgid matches, or returns zero if it
 * does not match. This is meant for use in code paths where the bit is expected
 * to be present and will be sufficient to protect against a short-term group
 * migration (e.g. takss and return from iocb).
 */
static inline ulong fd_get_running(int fd, uint desired_tgid)
{
  ulong ret = 0;
  uint old;

  /* TODO: may also be checked using an atomic double-load from a DWCAS
   * on compatible architectures, which wouldn't require to modify nor
   * restore the original value.
   */
  old = _HA_ATOMIC_ADD_FETCH(&fdtab[fd].refc_tgid, 0x10000);
  if (likely((old & 0xffff) == desired_tgid))
    ret = _HA_ATOMIC_LOAD(&fdtab[fd].running_mask);
  _HA_ATOMIC_SUB(&fdtab[fd].refc_tgid, 0x10000);
  return ret;
}

/* remove tid_bit from the fd's running mask and returns the value before the
 * atomic operation, so that the caller can know if it was present.
 */
static inline long fd_clr_running(int fd)
{
  return _HA_ATOMIC_FETCH_AND(&fdtab[fd].running_mask, ~ti->ltid_bit);
}

/* Prepares <fd> for being polled on all permitted threads of this group ID
 * (these will then be refined to only cover running ones).
*/
static inline void fd_insert(int fd, void *owner, void (*iocb)(int fd), int tgid, unsigned long thread_mask)
{
  extern void sock_conn_iocb(int);
  struct tgroup_info *tginfo = &ha_tgroup_info[tgid - 1];
  int newstate;

  /* conn_fd_handler should support edge-triggered FDs */
  newstate = 0;
  if ((global.tune.options & GTUNE_FD_ET) && iocb == sock_conn_iocb)
    newstate |= FD_ET_POSSIBLE;

  /* We must update fd_highest to reflect the highest known FD for this
   * thread. It's important to note that it's not necessarily the highest
   * FD the thread will see, it's the highest FD that was inserted by
   * this thread or by the main thread. The purpose is essentially to
   * let all threads know the highest known FD at boot, that will be
   * cloned into each thread, in order to limit the work range for init
   * functions such as fork_poller() and fd_reregister_all(). Keeping the
   * value thread-local substantially limits the cost, since after a few
   * thousand calls the value will just stop changing.
   */
  if (unlikely(fd > fd_highest))
    fd_highest = fd;

  /* This must never happen and would definitely indicate a bug, in
   * addition to overwriting some unexpected memory areas.
   */
  BUG_ON(fd < 0);
  BUG_ON(fd >= global.maxsock);
  BUG_ON(fdtab[fd].owner != NULL);
  BUG_ON(fdtab[fd].state != 0);
  BUG_ON(tgid < 1 || tgid > MAX_TGROUPS);

  thread_mask &= tginfo->threads_enabled;
  BUG_ON(thread_mask == 0);

  fd_claim_tgid(fd, tgid);

  BUG_ON(fdtab[fd].running_mask);

  fdtab[fd].owner = owner;
  fdtab[fd].iocb = iocb;
  fdtab[fd].state = newstate;
  fdtab[fd].thread_mask = thread_mask;

  /* just for debugging: how many times taken over since last fd_insert() */
  fdtab[fd].nb_takeover = 0;

  fd_drop_tgid(fd);

#ifdef DEBUG_FD
  fdtab[fd].event_count = 0;
#endif

  /* note: do not reset polled_mask here as it indicates which poller
   * still knows this FD from a possible previous round.
   */

  /* the two directions are ready until proven otherwise */
  fd_may_both(fd);
  _HA_ATOMIC_INC(&ha_used_fds);
}

/* 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)));
}

/* send a wake-up event to this thread, only if it's asleep and not notified yet */
static inline void wake_thread(int thr)
{
  struct thread_ctx *ctx = &ha_thread_ctx[thr];

  if ((_HA_ATOMIC_FETCH_OR(&ctx->flags, TH_FL_NOTIFIED) & (TH_FL_SLEEPING|TH_FL_NOTIFIED)) == TH_FL_SLEEPING) {
    char c = 'c';
    DISGUISE(write(poller_wr_pipe[thr], &c, 1));
  }
}


#endif /* _HAPROXY_FD_H */

/*
 * Local variables:
 *  c-indent-level: 8
 *  c-basic-offset: 8
 * End:
 */

Coverage Report

Created: 2026-09-03 06:33

Line	Count	Source
1		/*
2		* include/haproxy/fd.h
3		* File descriptors states - exported variables and functions
4		*
5		* Copyright (C) 2000-2020 Willy Tarreau - w@1wt.eu
6		*
7		* This library is free software; you can redistribute it and/or
8		* modify it under the terms of the GNU Lesser General Public
9		* License as published by the Free Software Foundation, version 2.1
10		* exclusively.
11		*
12		* This library is distributed in the hope that it will be useful,
13		* but WITHOUT ANY WARRANTY; without even the implied warranty of
14		* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15		* Lesser General Public License for more details.
16		*
17		* You should have received a copy of the GNU Lesser General Public
18		* License along with this library; if not, write to the Free Software
19		* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
20		*/
21
22		#ifndef _HAPROXY_FD_H
23		#define _HAPROXY_FD_H
24
25		#include <sys/time.h>
26		#include <sys/types.h>
27		#include <stdio.h>
28		#include <unistd.h>
29		#include <import/ist.h>
30		#include <haproxy/api.h>
31		#include <haproxy/atomic.h>
32		#include <haproxy/fd-t.h>
33		#include <haproxy/global.h>
34		#include <haproxy/thread.h>
35
36		/* public variables */
37
38		extern struct poller cur_poller; /* the current poller */
39		extern int nbpollers;