/*

 * Protocol registration functions.

 *

 * Copyright 2000-2012 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.

 *

 */

#include <unistd.h>

#include <sys/types.h>

#include <sys/socket.h>

#include <haproxy/api.h>

#include <haproxy/errors.h>

#include <haproxy/global.h>

#include <haproxy/list.h>

#include <haproxy/listener.h>

#include <haproxy/proto_quic.h>

#include <haproxy/protocol.h>

#include <haproxy/proxy.h>

#include <haproxy/quic_tune.h>

#include <haproxy/sock.h>

#include <haproxy/tools.h>

#include <haproxy/twork.h>

/* List head of all registered protocols */

static struct list protocols = LIST_HEAD_INIT(protocols);

struct protocol *__protocol_by_family[AF_CUST_MAX][PROTO_NUM_TYPES][2] __read_mostly = { };

const struct proto_fam *__proto_fam_by_family[AF_CUST_MAX] = { };

/* This is the global spinlock we may need to register/unregister listeners or

 * protocols. Its main purpose is in fact to serialize the rare stop/deinit()

 * phases.

 */

__decl_spinlock(proto_lock);

/* Registers the protocol <proto> */

void protocol_register(struct protocol *proto)

{

  int sock_family = proto->fam->sock_family;

  BUG_ON(sock_family < 0 || sock_family >= AF_CUST_MAX);

  BUG_ON(proto->proto_type >= PROTO_NUM_TYPES);

  LIST_INIT(&proto->receivers);

  proto->nb_receivers = 0;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  LIST_APPEND(&protocols, &proto->list);

  __protocol_by_family[sock_family]

                      [proto->proto_type]

                      [proto->xprt_type == PROTO_TYPE_DGRAM ||

                       proto->sock_prot == IPPROTO_MPTCP] = proto;

  __proto_fam_by_family[sock_family] = proto->fam;

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

}

/* Unregisters the protocol <proto>. Note that all listeners must have

 * previously been unbound.

 */

void protocol_unregister(struct protocol *proto)

{

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  LIST_DELETE(&proto->list);

  LIST_INIT(&proto->list);

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

}

/* clears flag <flag> on all protocols. */

void protocol_clrf_all(uint flag)

{

  struct protocol *proto;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list)

    proto->flags &= ~flag;

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

}

/* sets flag <flag> on all protocols. */

void protocol_setf_all(uint flag)

{

  struct protocol *proto;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list)

    proto->flags |= flag;

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

}

/* Checks if protocol <proto> supports PROTO_F flag <flag>. Returns zero if not,

 * non-zero if supported. It may return a cached value from a previous test,

 * and may run live tests then update the proto's flags to cache a result. It's

 * better to call it only if needed so that it doesn't result in modules being

 * loaded in case of a live test. It is only supposed to be used during boot.

 */

int protocol_supports_flag(struct protocol *proto, uint flag)

{

  if (flag == PROTO_F_REUSEPORT_SUPPORTED) {

    int ret = 0;

    /* check if the protocol supports SO_REUSEPORT */

    if (!(_HA_ATOMIC_LOAD(&proto->flags) & PROTO_F_REUSEPORT_SUPPORTED))

      return 0;

    /* TESTED is set, assume supported (live test already ran and passed) */

    if (_HA_ATOMIC_LOAD(&proto->flags) & PROTO_F_REUSEPORT_TESTED)

      return 1;

    /* run a live check */

    ret = _sock_supports_reuseport(proto->fam, proto->sock_type, proto->sock_prot);

    if (!ret)

      _HA_ATOMIC_AND(&proto->flags, ~PROTO_F_REUSEPORT_SUPPORTED);

    _HA_ATOMIC_OR(&proto->flags, PROTO_F_REUSEPORT_TESTED);

    return ret;

  }

  return 0;

}

#ifdef USE_QUIC

/* Return 1 if QUIC protocol may be bound, 0 if no, depending on the tuning

 * parameters.

 */

static inline int protocol_may_bind_quic(struct listener *l)

{

  return !(quic_tune.fe.opts & QUIC_TUNE_FE_LISTEN_OFF);

}

#endif

void protocol_init_rx_agents(void)

{

  struct protocol *proto;

  struct receiver *rx;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(rx, &proto->receivers, proto_list) {

      twork_init(&rx->agent.twk);

      rx->agent.want_state = RX_AGENT_ST_NONE;

      rx->agent.close_fd = -1;

      rx->agent.xfer_fd = -1;

      rx->agent.getsocks_fd = -1;

    }

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

}

/*

 * With per-thread-group FD tables, gives up the calling thread's group's

 * inherited copies of the receiver FDs it does not own: a receiver's FD is

 * only ever used from its owner group, and keeping the other groups' copies

 * alive would make every release require all groups' cooperation.

 */

void protocol_localize_rx_fds(void)

{

  struct protocol *proto;

  struct receiver *rx;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(rx, &proto->receivers, proto_list) {

      int fd = rx->fd;

      if (fd < 0 || rx_owner_tgid(rx) == tgid)

        continue;

      fdtab[fd].owner = NULL;

      fdtab[fd].state = 0;

      fdtab[fd].thread_mask = 0;

      fdtab[fd].update_mask = 0;

      fdtab[fd].running_mask = 0;

      HA_ATOMIC_STORE(&fdtab[fd].refc_tgid, 0);

      close(fd);

    }

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

}

static uint getsocks_busy;

/*

 * Collect file descriptor from other thread groups, if they have their

 * own file descriptors table.

 */

int protocol_getsocks_foreign_fds(struct receiver ***orxs, int **ofds)

{

  struct protocol *proto;

  struct receiver *rx;

  struct receiver **rxs;

  int *fds;

  int nb, filled, done, i;

  uint idle = 0;

  *orxs = NULL;

  *ofds = NULL;

  if (MAX_TGROUPS < 2 || !(global.tune.options & GTUNE_NO_TG_FD_SHARING) || global.nbtgroups < 2)

    return 0;

  /* a concurrent collection would overwrite the per-receiver slots, and

   * two of them blocking their groups' runner threads would deadlock.

   */

  if (!HA_ATOMIC_CAS(&getsocks_busy, &idle, 1)) {

    ha_warning("_getsocks: a collection is already in progress, giving up.\n");

    return -1;

  }

  /* first pass: count the candidates */

  nb = 0;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(rx, &proto->receivers, proto_list) {

      if (rx->fd < 0 || rx_owner_tgid(rx) == tgid)

        continue;

      if (!(ha_tgroup_ctx[rx_owner_tgid(rx) - 1].fdtab[rx->fd].state & FD_EXPORTED))

        continue;

      nb++;

    }

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

  if (!nb) {

    HA_ATOMIC_STORE(&getsocks_busy, 0);

    return 0;

  }

  rxs = calloc(nb, sizeof(*rxs));

  fds = calloc(nb, sizeof(*fds));

  if (!rxs || !fds) {

    free(rxs);

    free(fds);

    HA_ATOMIC_STORE(&getsocks_busy, 0);

    ha_warning("Failed to allocate memory to transfer other groups' sockets\n");

    return -1;

  }

  /* second pass: arm the requests. A copy left over by a previous

   * timed-out collection is closed on the way.

   */

  filled = 0;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(rx, &proto->receivers, proto_list) {

      int old;

      if (filled >= nb)

        break;

      if (rx->fd < 0 || rx_owner_tgid(rx) == tgid)

        continue;

      if (!(ha_tgroup_ctx[rx_owner_tgid(rx) - 1].fdtab[rx->fd].state & FD_EXPORTED))

        continue;

      old = rx->agent.getsocks_fd;

      if (old >= 0)

        close(old);

      HA_ATOMIC_STORE(&rx->agent.getsocks_fd, -1);

      rxs[filled++] = rx;

      rx_agent_getsocks_request(rx, tgid);

    }

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

  nb = filled;

  for (i = 0; i < 50; i++) {

    rx_xfer_drain(tgid - 1);

    done = 1;

    for (filled = 0; filled < nb; filled++) {

      if (HA_ATOMIC_LOAD(&rxs[filled]->agent.getsocks_fd) < 0)

        done = 0;

    }

    if (done)

      break;

    /* XXX: we're waiting for other thread groups to send their

     * fds, let's sleep for a bit.

     */

    usleep(10000);

  }

  done = 0;

  for (i = 0; i < nb; i++) {

    int f = HA_ATOMIC_LOAD(&rxs[i]->agent.getsocks_fd);

    if (f < 0) {

      ha_warning("_getsocks: could not get a copy of a listener FD from thread group %u, the new process will have to bind it itself.\n",

                 rx_owner_tgid(rxs[i]));

      continue;

    }

    HA_ATOMIC_STORE(&rxs[i]->agent.getsocks_fd, -1);

    rxs[done] = rxs[i];

    fds[done] = f;

    done++;

  }

  HA_ATOMIC_STORE(&getsocks_busy, 0);

  if (!done) {

    free(rxs);

    free(fds);

    return 0;

  }

  *orxs = rxs;

  *ofds = fds;

  return done;

}

/* binds all listeners of all registered protocols. Returns a composition

 * of ERR_NONE, ERR_RETRYABLE, ERR_FATAL.

 */

int protocol_bind_all(int verbose)

{

  struct protocol *proto;

  struct listener *listener;

  struct receiver *receiver;

  char msg[1000];

  char *errmsg;

  int err, lerr;

  err = 0;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(receiver, &proto->receivers, proto_list) {

      listener = LIST_ELEM(receiver, struct listener *, rx);

#ifdef USE_QUIC

      if ((proto == &proto_quic4 || proto == &proto_quic6) &&

          !protocol_may_bind_quic(listener))

        continue;

#endif

      lerr = proto->fam->bind(receiver, &errmsg);

      err |= lerr;

      /* errors are reported if <verbose> is set or if they are fatal */

      if (verbose || (lerr & (ERR_FATAL | ERR_ABORT))) {

        struct proxy *px = listener->bind_conf->frontend;

        if (lerr & ERR_ALERT)

          ha_alert("Binding [%s:%d] for %s %s: protocol %s: %s.\n",

                   listener->bind_conf->file, listener->bind_conf->line,

             proxy_type_str(px), px->id, proto->name, errmsg);

        else if (lerr & ERR_WARN)

          ha_warning("Binding [%s:%d] for %s %s: protocol %s: %s.\n",

                     listener->bind_conf->file, listener->bind_conf->line,

               proxy_type_str(px), px->id, proto->name, errmsg);

      }

      if (lerr != ERR_NONE)

        ha_free(&errmsg);

      if (lerr & ERR_ABORT)

        break;

      if (lerr & ~ERR_WARN)

        continue;

      /* for now there's still always a listening function */

      BUG_ON(!proto->listen);

      lerr = proto->listen(listener, msg, sizeof(msg));

      err |= lerr;

      if (verbose || (lerr & (ERR_FATAL | ERR_ABORT))) {

        struct proxy *px = listener->bind_conf->frontend;

        if (lerr & ERR_ALERT)

          ha_alert("Starting [%s:%d] for %s %s: protocol %s: %s.\n",

                   listener->bind_conf->file, listener->bind_conf->line,

             proxy_type_str(px), px->id, proto->name, msg);

        else if (lerr & ERR_WARN)

          ha_warning("Starting [%s:%d] for %s %s: protocol %s: %s.\n",

                     listener->bind_conf->file, listener->bind_conf->line,

               proxy_type_str(px), px->id, proto->name, msg);

      }

      if (lerr & ERR_ABORT)

        break;

    }

    if (err & ERR_ABORT)

      break;

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

  return err;

}

/* unbinds all listeners of all registered protocols. They are also closed.

 * This must be performed before calling exit() in order to get a chance to

 * remove file-system based sockets and pipes.

 * Returns a composition of ERR_NONE, ERR_RETRYABLE, ERR_FATAL, ERR_ABORT.

 */

int protocol_unbind_all(void)

{

  struct protocol *proto;

  struct listener *listener;

  int err;

  err = 0;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(listener, &proto->receivers, rx.proto_list)

      unbind_listener(listener);

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

  return err;

}

/* stops all listeners of all registered protocols. This will normally catch

 * every single listener, all protocols included. This is to be used during

 * soft_stop() only. It does not return any error.

 */

void protocol_stop_now(void)

{

  struct protocol *proto;

  struct listener *listener, *lback;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry_safe(listener, lback, &proto->receivers, rx.proto_list)

      stop_listener(listener, 0, 1, 0);

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

}

/* suspends all listeners of all registered protocols. This is typically

 * used on SIG_TTOU to release all listening sockets for the time needed to

 * try to bind a new process. The listeners enter LI_PAUSED or LI_ASSIGNED.

 * It returns ERR_NONE, with ERR_FATAL on failure.

 */

int protocol_pause_all(void)

{

  struct protocol *proto;

  struct listener *listener;

  int err;

  err = 0;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(listener, &proto->receivers, rx.proto_list)

      if (!suspend_listener(listener, 0, 0))

        err |= ERR_FATAL;

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

  return err;

}

/* resumes all listeners of all registered protocols. This is typically used on

 * SIG_TTIN to re-enable listening sockets after a new process failed to bind.

 * The listeners switch to LI_READY/LI_FULL. It returns ERR_NONE, with ERR_FATAL

 * on failure.

 */

int protocol_resume_all(void)

{

  struct protocol *proto;

  struct listener *listener;

  int err;

  err = 0;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(listener, &proto->receivers, rx.proto_list)

      if (!resume_listener(listener, 0, 0))

        err |= ERR_FATAL;

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

  return err;

}

/* enables all listeners of all registered protocols. This is intended to be

 * used after a fork() to enable reading on all file descriptors. Returns ERR_NONE.

 */

int protocol_enable_all(void)

{

  struct protocol *proto;

  struct listener *listener;

  HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);

  list_for_each_entry(proto, &protocols, list) {

    list_for_each_entry(listener, &proto->receivers, rx.proto_list)

      enable_listener(listener);

  }

  HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);

  return ERR_NONE;

}

/*

 * Local variables:

 *  c-indent-level: 8

 *  c-basic-offset: 8

 * End:

 */