ullong maxconn_reached = 0;
__decl_thread(static HA_RWLOCK_T global_listener_rwlock);
+static void rx_agent_schedule_op(struct receiver *rx, uint op);
+
/* listener status for stats */
const char* li_status_st[LI_STATE_COUNT] = {
[LI_STATUS_WAITING] = "WAITING",
*/
void enable_listener(struct listener *listener)
{
+ if (tg_agents_enabled && rx_owner_tgid(&listener->rx) != tgid) {
+ rx_agent_schedule_op(&listener->rx, RX_AGENT_OP_ENABLE);
+ return;
+ }
+
HA_RWLOCK_WRLOCK(LISTENER_LOCK, &listener->lock);
/* If this listener is supposed to be only in the master, close it in
if (!(l->flags & LI_F_FINALIZED) || l->state <= LI_PAUSED)
goto end;
+ if (tg_agents_enabled && rx_owner_tgid(&l->rx) != tgid) {
+ rx_agent_schedule_op(&l->rx, RX_AGENT_OP_SUSPEND);
+ goto end;
+ }
+
if (l->rx.proto->suspend) {
ret = l->rx.proto->suspend(l);
/* if the suspend() fails, we don't want to change the
if (!(l->flags & LI_F_FINALIZED) || l->state == LI_READY)
goto end;
+ if (tg_agents_enabled && l->state == LI_ASSIGNED &&
+ (l->rx.flags & RX_F_MUST_DUP)) {
+ if (l->rx.agent.xfer_fd < 0) {
+ rx_agent_schedule_op(l->rx.shard_info->ref, RX_AGENT_OP_RESUME);
+ goto end;
+ }
+ }
+ else if (tg_agents_enabled && rx_owner_tgid(&l->rx) != tgid) {
+ /* all FD operations, including a possible rebind, must act on
+ * the owner group's kernel and fdtab tables: delegate to its
+ * agent and report success, failures will be handled there.
+ */
+ rx_agent_schedule_op(&l->rx, RX_AGENT_OP_RESUME);
+ goto end;
+ }
+
if (l->rx.proto->resume) {
ret = l->rx.proto->resume(l);
if (!ret)
}
+/*
+ * Per-thread-group agents, only set up when FD tables are not shared between
+ * thread groups. Each thread group runs one agent tasklet on its first thread,
+ * to which any thread may post the operations that can only be performed from
+ * inside that group.
+ */
+struct tg_agent {
+ struct tasklet *tl; /* runs on the group's first thread */
+ struct mt_list ops; /* receivers with pending operations */
+ int xfer_sock[2]; /* [0]=any sender, [1]=this group's agent */
+};
+
+static struct tg_agent tg_agents[MAX_TGROUPS];
+int tg_agents_enabled = 0;
+
+enum rx_xfer_op {
+ RX_XFER_OP_REBIND = 0,
+};
+
+struct rx_xfer_msg {
+ struct receiver *rx;
+ enum rx_xfer_op op;
+};
+
+static void rx_xfer_send_members(struct receiver *ref);
+static int rx_xfer_send_fd(uint grp, struct receiver *rx, enum rx_xfer_op op, int fd);
+static void rx_xfer_drain(uint grp);
+
+/* Queues receiver <rx> to 1-based group <grp>'s agent and wakes it. The
+ * operation fields must be set before calling this, see struct tg_agent.
+ */
+static void rx_agent_post(struct receiver *rx, uint grp)
+{
+ if (MT_LIST_TRY_APPEND(&tg_agents[grp - 1].ops, &rx->agent.link.list))
+ tasklet_wakeup(tg_agents[grp - 1].tl);
+}
+
+static void rx_agent_process_one(struct receiver *rx)
+{
+ int fd = HA_ATOMIC_XCHG(&rx->agent.close_fd, -1);
+ uint ops;
+
+ if (fd >= 0)
+ fd_delete(fd);
+
+ ops = HA_ATOMIC_XCHG(&rx->agent.ops, 0);
+ if (ops) {
+ struct listener *l = LIST_ELEM(rx, struct listener *, rx);
+
+ if (ops & RX_AGENT_OP_SUSPEND)
+ suspend_listener(l, 0, 0);
+
+ if (ops & RX_AGENT_OP_RESUME) {
+ resume_listener(l, 0, 0);
+
+ /*
+ * if this is the reference of a shard spanning several
+ * groups, pass a copy of the (re)bound FD to the groups
+ * of the members that still need one, their own group
+ * will finish their resume with it.
+ */
+ if ((rx->flags & RX_F_BOUND) && rx->shard_info &&
+ rx->shard_info->ref == rx)
+ rx_xfer_send_members(rx);
+ }
+
+ if (ops & RX_AGENT_OP_ENABLE)
+ enable_listener(l);
+ }
+}
+
+static struct task *tg_agent_process(struct task *t, void *context, unsigned int state)
+{
+ uint grp = (uint)(ulong)context;
+ struct rx_agent_link *lnk;
+
+ while ((lnk = MT_LIST_POP(&tg_agents[grp].ops, struct rx_agent_link *, list)))
+ rx_agent_process_one(lnk->rx);
+
+ rx_xfer_drain(grp);
+ return t;
+}
+
+void rx_agent_close(struct receiver *rx)
+{
+ HA_ATOMIC_STORE(&rx->agent.close_fd, rx->fd);
+ rx_agent_post(rx, rx_owner_tgid(rx));
+}
+
+static void rx_agent_schedule_op(struct receiver *rx, uint op)
+{
+ HA_ATOMIC_OR(&rx->agent.ops, op);
+ rx_agent_post(rx, rx_owner_tgid(rx));
+}
+
+static void rx_xfer_send_members(struct receiver *ref)
+{
+ const struct shard_info *si = ref->shard_info;
+ uint i;
+
+ for (i = 0; i < si->nbgroups; i++) {
+ struct receiver *rx = si->members[i];
+ uint grp;
+
+ if (rx == ref || (rx->flags & RX_F_BOUND))
+ continue;
+
+ grp = rx_owner_tgid(rx);
+ rx_xfer_send_fd(grp, rx, RX_XFER_OP_REBIND, ref->fd);
+ }
+}
+
+static int rx_xfer_send_fd(uint grp, struct receiver *rx, enum rx_xfer_op op, int fd)
+{
+ struct rx_xfer_msg xmsg;
+ struct msghdr msg;
+ struct iovec iov;
+ struct cmsghdr *cmsg;
+ union {
+ char buf[CMSG_SPACE(sizeof(int))];
+ struct cmsghdr align;
+ } u;
+
+ xmsg.rx = rx;
+ xmsg.op = op;
+
+ memset(&msg, 0, sizeof(msg));
+ memset(&u, 0, sizeof(u));
+ iov.iov_base = &xmsg;
+ iov.iov_len = sizeof(xmsg);
+ msg.msg_iov = &iov;
+ msg.msg_iovlen = 1;
+ msg.msg_control = u.buf;
+ msg.msg_controllen = sizeof(u.buf);
+ cmsg = CMSG_FIRSTHDR(&msg);
+ cmsg->cmsg_level = SOL_SOCKET;
+ cmsg->cmsg_type = SCM_RIGHTS;
+ cmsg->cmsg_len = CMSG_LEN(sizeof(int));
+ memcpy(CMSG_DATA(cmsg), &fd, sizeof(int));
+
+ if (sendmsg(tg_agents[grp - 1].xfer_sock[0], &msg, MSG_DONTWAIT) < 0) {
+ ha_warning("Failed to transfer a listener FD copy to thread group %u (%s).\n",
+ grp, strerror(errno));
+ return 0;
+ }
+ tasklet_wakeup(tg_agents[grp - 1].tl);
+ return 1;
+}
+
+static void rx_xfer_drain(uint grp)
+{
+ while (1) {
+ struct rx_xfer_msg xmsg;
+ struct receiver *rx;
+ struct listener *l;
+ struct msghdr msg;
+ struct iovec iov;
+ struct cmsghdr *cmsg;
+ union {
+ char buf[CMSG_SPACE(sizeof(int))];
+ struct cmsghdr align;
+ } u;
+ ssize_t ret;
+ int fd = -1;
+
+ memset(&msg, 0, sizeof(msg));
+ iov.iov_base = &xmsg;
+ iov.iov_len = sizeof(xmsg);
+ msg.msg_iov = &iov;
+ msg.msg_iovlen = 1;
+ msg.msg_control = u.buf;
+ msg.msg_controllen = sizeof(u.buf);
+
+ ret = recvmsg(tg_agents[grp].xfer_sock[1], &msg, MSG_DONTWAIT | MSG_CMSG_CLOEXEC);
+ if (ret <= 0)
+ break;
+
+ cmsg = CMSG_FIRSTHDR(&msg);
+ if (cmsg && cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS)
+ memcpy(&fd, CMSG_DATA(cmsg), sizeof(int));
+
+ if (ret != sizeof(xmsg) || fd < 0) {
+ if (fd >= 0)
+ close(fd);
+ continue;
+ }
+ rx = xmsg.rx;
+
+ switch (xmsg.op) {
+ case RX_XFER_OP_REBIND:
+ rx->agent.xfer_fd = fd;
+ l = LIST_ELEM(rx, struct listener *, rx);
+ resume_listener(l, 0, 0);
+ if (rx->agent.xfer_fd >= 0) {
+ close(rx->agent.xfer_fd);
+ rx->agent.xfer_fd = -1;
+ }
+ break;
+
+ default:
+ close(fd);
+ break;
+ }
+ }
+}
+
+int rx_agent_init(void)
+{
+ uint i;
+
+ for (i = 0; i < global.nbtgroups; i++) {
+ MT_LIST_INIT(&tg_agents[i].ops);
+ if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, tg_agents[i].xfer_sock) < 0)
+ return 0;
+ fd_set_nonblock(tg_agents[i].xfer_sock[1]);
+ tg_agents[i].tl = tasklet_new();
+ if (!tg_agents[i].tl)
+ return 0;
+ tg_agents[i].tl->process = tg_agent_process;
+ tg_agents[i].tl->context = (void *)(ulong)i;
+ tg_agents[i].tl->tid = ha_tgroup_info[i].base;
+ }
+ protocol_init_rx_agents();
+ tg_agents_enabled = 1;
+ return 1;
+}
+
/* default function used to unbind a listener. This is for use by standard
* protocols working on top of accepted sockets. The receiver's rx_unbind()
* will automatically be used after the listener is disabled if the socket is
*
*/
+#include <unistd.h>
+
#include <sys/types.h>
#include <sys/socket.h>
}
#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) {
+ MT_LIST_INIT(&rx->agent.link.list);
+ rx->agent.link.rx = rx;
+ rx->agent.close_fd = -1;
+ rx->agent.xfer_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);
+}
+
/* binds all listeners of all registered protocols. Returns a composition
* of ERR_NONE, ERR_RETRYABLE, ERR_FATAL.
*/
projects / thirdparty / haproxy.git / commitdiff
