}
}
+/*
+ * Send one FD, with its relevant informations.
+ */
+static int _getsocks_send_one(int sock, int send_fd, const struct receiver *rx,
+ struct msghdr *msghdr, unsigned char *tmpbuf,
+ int *tmpfd, int *nb_queued, int *curoff)
+{
+ const char *ns_name, *if_name;
+ unsigned char ns_nlen, if_nlen;
+
+ ns_name = if_name = "";
+ ns_nlen = if_nlen = 0;
+
+ /* for now we can only retrieve namespaces and interfaces from
+ * pure listeners.
+ */
+ if (rx && rx->iocb == sock_accept_iocb) {
+ if (rx->settings->interface) {
+ if_name = rx->settings->interface;
+ if_nlen = strlen(if_name);
+ }
+
+#ifdef USE_NS
+ if (rx->settings->netns) {
+ ns_name = rx->settings->netns->node.key;
+ ns_nlen = rx->settings->netns->name_len;
+ }
+#endif
+ }
+
+ /* put the FD into the CMSG_DATA */
+ tmpfd[(*nb_queued)++] = send_fd;
+
+ /* first block is <ns_name_len> <ns_name> */
+ tmpbuf[(*curoff)++] = ns_nlen;
+ if (ns_nlen)
+ memcpy(tmpbuf + *curoff, ns_name, ns_nlen);
+ *curoff += ns_nlen;
+
+ /* second block is <if_name_len> <if_name> */
+ tmpbuf[(*curoff)++] = if_nlen;
+ if (if_nlen)
+ memcpy(tmpbuf + *curoff, if_name, if_nlen);
+ *curoff += if_nlen;
+
+ /* we used to send the listener options here before 2.3 */
+ memset(tmpbuf + *curoff, 0, sizeof(int));
+ *curoff += sizeof(int);
+
+ /* there's a limit to how many FDs may be sent at once */
+ if (*nb_queued == MAX_SEND_FD) {
+ int ack, ret;
+
+ msghdr->msg_iov->iov_len = *curoff;
+ if (sendmsg(sock, msghdr, 0) != *curoff) {
+ ha_warning("Failed to transfer sockets\n");
+ return 0;
+ }
+
+ /* Wait for an ack */
+ do {
+ ret = recv(sock, &ack, sizeof(ack), 0);
+ } while (ret == -1 && errno == EINTR);
+
+ if (ret <= 0) {
+ ha_warning("Unexpected error while transferring sockets\n");
+ return 0;
+ }
+ *curoff = 0;
+ *nb_queued = 0;
+ }
+ return 1;
+}
+
/* Send all the bound sockets, always returns 1 */
static int _getsocks(char **args, char *payload, struct appctx *appctx, void *private)
{
struct msghdr msghdr;
struct iovec iov;
struct timeval tv = { .tv_sec = 1, .tv_usec = 0 };
- const char *ns_name, *if_name;
- unsigned char ns_nlen, if_nlen;
+ struct receiver **foreign_rxs = NULL;
+ int *foreign_fds = NULL;
+ int nb_foreign = 0;
+ int foreign_idx;
int nb_queued;
int cur_fd = 0;
int *tmpfd;
if (!(strm_li(s)->bind_conf->level & ACCESS_FD_LISTENERS))
goto out;
memset(&msghdr, 0, sizeof(msghdr));
+
+ /*
+ * Collect file descriptors from other thread groups, if they have
+ * a separate file descriptor table.
+ */
+ nb_foreign = protocol_getsocks_foreign_fds(&foreign_rxs, &foreign_fds);
+ if (nb_foreign < 0) {
+ ha_warning("Failed to allocate memory to transfer other groups' sockets\n");
+ goto out;
+ }
+
/*
* First, calculates the total number of FD, so that we can let
* the caller know how much it should expect.
for (cur_fd = 0;cur_fd < global.maxsock; cur_fd++)
tot_fd_nb += !!(fdtab[cur_fd].state & FD_EXPORTED);
+ tot_fd_nb += nb_foreign;
+
if (tot_fd_nb == 0) {
if (already_sent)
ha_warning("_getsocks: attempt to get sockets but they were already sent and closed in this process!\n");
nb_queued = 0;
iov.iov_base = tmpbuf;
for (cur_fd = 0; cur_fd < global.maxsock; cur_fd++) {
+ const struct receiver *rx = NULL;
+
if (!(fdtab[cur_fd].state & FD_EXPORTED))
continue;
/* this FD is now shared between processes */
HA_ATOMIC_OR(&fdtab[cur_fd].state, FD_CLONED);
- ns_name = if_name = "";
- ns_nlen = if_nlen = 0;
+ if (fdtab[cur_fd].iocb == sock_accept_iocb)
+ rx = &((const struct listener *)fdtab[cur_fd].owner)->rx;
- /* for now we can only retrieve namespaces and interfaces from
- * pure listeners.
- */
- if (fdtab[cur_fd].iocb == sock_accept_iocb) {
- const struct listener *l = fdtab[cur_fd].owner;
-
- if (l->rx.settings->interface) {
- if_name = l->rx.settings->interface;
- if_nlen = strlen(if_name);
- }
-
-#ifdef USE_NS
- if (l->rx.settings->netns) {
- ns_name = l->rx.settings->netns->node.key;
- ns_nlen = l->rx.settings->netns->name_len;
- }
-#endif
- }
-
- /* put the FD into the CMSG_DATA */
- tmpfd[nb_queued++] = cur_fd;
-
- /* first block is <ns_name_len> <ns_name> */
- tmpbuf[curoff++] = ns_nlen;
- if (ns_nlen)
- memcpy(tmpbuf + curoff, ns_name, ns_nlen);
- curoff += ns_nlen;
-
- /* second block is <if_name_len> <if_name> */
- tmpbuf[curoff++] = if_nlen;
- if (if_nlen)
- memcpy(tmpbuf + curoff, if_name, if_nlen);
- curoff += if_nlen;
-
- /* we used to send the listener options here before 2.3 */
- memset(tmpbuf + curoff, 0, sizeof(int));
- curoff += sizeof(int);
-
- /* there's a limit to how many FDs may be sent at once */
- if (nb_queued == MAX_SEND_FD) {
- iov.iov_len = curoff;
- if (sendmsg(fd, &msghdr, 0) != curoff) {
- ha_warning("Failed to transfer sockets\n");
- goto out;
- }
-
- /* Wait for an ack */
- do {
- ret = recv(fd, &tot_fd_nb, sizeof(tot_fd_nb), 0);
- } while (ret == -1 && errno == EINTR);
+ if (!_getsocks_send_one(fd, cur_fd, rx, &msghdr, tmpbuf,
+ tmpfd, &nb_queued, &curoff))
+ goto out;
+ }
- if (ret <= 0) {
- ha_warning("Unexpected error while transferring sockets\n");
- goto out;
- }
- curoff = 0;
- nb_queued = 0;
- }
+ for (foreign_idx = 0; foreign_idx < nb_foreign; foreign_idx++) {
+ if (!_getsocks_send_one(fd, foreign_fds[foreign_idx],
+ foreign_rxs[foreign_idx], &msghdr,
+ tmpbuf, tmpfd, &nb_queued, &curoff))
+ goto out;
}
already_sent = 1;
}
out:
+ for (foreign_idx = 0; foreign_idx < nb_foreign; foreign_idx++)
+ close(foreign_fds[foreign_idx]);
+ free(foreign_rxs);
+ free(foreign_fds);
if (fd >= 0 && old_fcntl >= 0 && fcntl(fd, F_SETFL, old_fcntl) == -1)
ha_warning("Cannot make the unix socket non-blocking\n");
applet_set_eoi(appctx);
enum rx_xfer_op {
RX_XFER_OP_REBIND = 0,
+ RX_XFER_OP_GETSOCKS,
};
struct rx_xfer_msg {
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_process_one(struct receiver *rx)
{
int fd = HA_ATOMIC_XCHG(&rx->agent.close_fd, -1);
+ uint dest;
uint ops;
if (fd >= 0)
if (ops & RX_AGENT_OP_ENABLE)
enable_listener(l);
}
+
+ dest = HA_ATOMIC_XCHG(&rx->agent.getsocks_grp, 0);
+ if (dest && rx->fd >= 0) {
+ HA_ATOMIC_OR(&fdtab[rx->fd].state, FD_CLONED);
+ rx_xfer_send_fd(dest, rx, RX_XFER_OP_GETSOCKS, rx->fd);
+ }
}
static struct task *tg_agent_process(struct task *t, void *context, unsigned int state)
rx_agent_post(rx, rx_owner_tgid(rx));
}
+void rx_agent_getsocks_request(struct receiver *rx, uint dest_grp)
+{
+ HA_ATOMIC_STORE(&rx->agent.getsocks_grp, dest_grp);
+ 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;
return 1;
}
-static void rx_xfer_drain(uint grp)
+void rx_xfer_drain(uint grp)
{
while (1) {
struct rx_xfer_msg xmsg;
}
break;
+ case RX_XFER_OP_GETSOCKS:
+ HA_ATOMIC_STORE(&rx->agent.getsocks_fd, fd);
+ break;
+
default:
close(fd);
break;
rx->agent.link.rx = rx;
rx->agent.close_fd = -1;
rx->agent.xfer_fd = -1;
+ rx->agent.getsocks_fd = -1;
}
}
HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);
HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);
}
+/*
+ * 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;
+
+ *orxs = NULL;
+ *ofds = NULL;
+
+ if (!(global.tune.options & GTUNE_NO_TG_FD_SHARING) || global.nbtgroups < 2)
+ return 0;
+
+ /* 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)
+ return 0;
+
+ rxs = calloc(nb, sizeof(*rxs));
+ fds = calloc(nb, sizeof(*fds));
+ if (!rxs || !fds) {
+ free(rxs);
+ free(fds);
+ 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++;
+ }
+
+ 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.
*/
projects / thirdparty / haproxy.git / commitdiff
