X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fht-cleanup.cpp;fp=src%2Fbin%2Flttng-sessiond%2Fht-cleanup.cpp;h=050f18e2bc38a5fb0e65f1044c687b516b0f240a;hb=7966af5763c4aaca39df9bbfa9277ff15715c720;hp=0000000000000000000000000000000000000000;hpb=3a5f70173aa04d11ccb22694d5d31a702cad33ab;p=lttng-tools.git

diff --git a/src/bin/lttng-sessiond/ht-cleanup.cpp b/src/bin/lttng-sessiond/ht-cleanup.cpp
new file mode 100644
index 000000000..050f18e2b
--- /dev/null
+++ b/src/bin/lttng-sessiond/ht-cleanup.cpp
@@ -0,0 +1,267 @@
+/*
+ * Copyright (C) 2013 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * SPDX-License-Identifier: GPL-2.0-only
+ *
+ */
+
+#define _LGPL_SOURCE
+
+#include <common/hashtable/hashtable.h>
+#include <common/common.h>
+#include <common/utils.h>
+#include <pthread.h>
+
+#include "lttng-sessiond.h"
+#include "health-sessiond.h"
+#include "testpoint.h"
+#include "utils.h"
+#include "ht-cleanup.h"
+
+static int ht_cleanup_quit_pipe[2] = { -1, -1 };
+
+/*
+ * Check if the ht_cleanup thread quit pipe was triggered.
+ *
+ * Return true if it was triggered else false;
+ */
+static bool check_quit_pipe(int fd, uint32_t events)
+{
+	return (fd == ht_cleanup_quit_pipe[0] && (events & LPOLLIN));
+}
+
+static int init_pipe(int *pipe_fds)
+{
+	int ret, i;
+
+	ret = pipe(pipe_fds);
+	if (ret < 0) {
+		PERROR("ht_cleanup thread quit pipe");
+		goto error;
+	}
+
+	for (i = 0; i < 2; i++) {
+		ret = fcntl(pipe_fds[i], F_SETFD, FD_CLOEXEC);
+		if (ret < 0) {
+			PERROR("fcntl ht_cleanup_quit_pipe");
+			goto error;
+		}
+	}
+error:
+	return ret;
+}
+
+/*
+ * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
+ */
+static int set_pollset(struct lttng_poll_event *events, size_t size)
+{
+	int ret;
+
+	ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
+	if (ret < 0) {
+		goto error;
+	}
+
+	ret = lttng_poll_add(events, ht_cleanup_quit_pipe[0],
+			LPOLLIN | LPOLLERR);
+	if (ret < 0) {
+		goto error;
+	}
+
+	ret = lttng_poll_add(events, the_ht_cleanup_pipe[0], LPOLLIN | LPOLLERR);
+	if (ret < 0) {
+		DBG("lttng_poll_add error %d.", ret);
+		goto error;
+	}
+
+	return 0;
+
+error:
+	return ret;
+}
+
+static void cleanup_ht_cleanup_thread(void *data)
+{
+	utils_close_pipe(ht_cleanup_quit_pipe);
+	utils_close_pipe(the_ht_cleanup_pipe);
+}
+
+static void *thread_ht_cleanup(void *data)
+{
+	int ret, i, pollfd, err = -1;
+	ssize_t size_ret;
+	uint32_t revents, nb_fd;
+	struct lttng_poll_event events;
+
+	DBG("startup.");
+
+	rcu_register_thread();
+	rcu_thread_online();
+
+	health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_HT_CLEANUP);
+
+	if (testpoint(sessiond_thread_ht_cleanup)) {
+		DBG("testpoint.");
+		goto error_testpoint;
+	}
+
+	health_code_update();
+
+	ret = set_pollset(&events, 2);
+	if (ret < 0) {
+		DBG("sessiond_set_ht_cleanup_thread_pollset error %d.", ret);
+		goto error_poll_create;
+	}
+
+	health_code_update();
+
+	while (1) {
+	restart:
+		DBG3("Polling.");
+		health_poll_entry();
+		ret = lttng_poll_wait(&events, -1);
+		DBG3("Returning from poll on %d fds.",
+			LTTNG_POLL_GETNB(&events));
+		health_poll_exit();
+		if (ret < 0) {
+			/*
+			 * Restart interrupted system call.
+			 */
+			if (errno == EINTR) {
+				continue;
+			}
+			goto error;
+		}
+
+		nb_fd = ret;
+		for (i = 0; i < nb_fd; i++) {
+			struct lttng_ht *ht;
+
+			health_code_update();
+
+			/* Fetch once the poll data */
+			revents = LTTNG_POLL_GETEV(&events, i);
+			pollfd = LTTNG_POLL_GETFD(&events, i);
+
+			if (pollfd != the_ht_cleanup_pipe[0]) {
+				continue;
+			}
+
+			if (revents & LPOLLIN) {
+				/* Get socket from dispatch thread. */
+				size_ret = lttng_read(the_ht_cleanup_pipe[0],
+						&ht, sizeof(ht));
+				if (size_ret < sizeof(ht)) {
+					PERROR("ht cleanup notify pipe");
+					goto error;
+				}
+				health_code_update();
+				/*
+				 * The whole point of this thread is to call
+				 * lttng_ht_destroy from a context that is NOT:
+				 * 1) a read-side RCU lock,
+				 * 2) a call_rcu thread.
+				 */
+				lttng_ht_destroy(ht);
+
+				health_code_update();
+
+				/*
+				 * Ensure that we never process the quit pipe
+				 * event while there is still data available
+				 * on the ht clean pipe.
+				 */
+				goto restart;
+			} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+				ERR("ht cleanup pipe error");
+				goto error;
+			} else {
+				ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+				goto error;
+			}
+		}
+
+		for (i = 0; i < nb_fd; i++) {
+			health_code_update();
+
+			/* Fetch once the poll data */
+			revents = LTTNG_POLL_GETEV(&events, i);
+			pollfd = LTTNG_POLL_GETFD(&events, i);
+
+			if (!revents) {
+				/* No activity for this FD (poll implementation). */
+				continue;
+			}
+
+			if (pollfd == the_ht_cleanup_pipe[0]) {
+				continue;
+			}
+
+			/* Thread quit pipe has been closed. Killing thread. */
+			ret = check_quit_pipe(pollfd, revents);
+			if (ret) {
+				err = 0;
+				DBG("[ht-cleanup] quit.");
+				goto exit;
+			}
+		}
+	}
+
+exit:
+error:
+	lttng_poll_clean(&events);
+error_poll_create:
+error_testpoint:
+	DBG("[ht-cleanup] Thread terminates.");
+	if (err) {
+		health_error();
+		ERR("Health error occurred in %s", __func__);
+	}
+	health_unregister(the_health_sessiond);
+	rcu_thread_offline();
+	rcu_unregister_thread();
+	return NULL;
+}
+
+static bool shutdown_ht_cleanup_thread(void *data)
+{
+	int ret;
+
+	ret = notify_thread_pipe(ht_cleanup_quit_pipe[1]);
+	if (ret < 0) {
+		ERR("write error on ht_cleanup quit pipe");
+		goto end;
+	}
+end:
+	return ret;
+}
+
+struct lttng_thread *launch_ht_cleanup_thread(void)
+{
+	int ret;
+	struct lttng_thread *thread;
+
+	ret = init_pipe(the_ht_cleanup_pipe);
+	if (ret) {
+		goto error;
+	}
+
+	ret = init_pipe(ht_cleanup_quit_pipe);
+	if (ret) {
+		goto error;
+	}
+
+	thread = lttng_thread_create("HT cleanup",
+			thread_ht_cleanup,
+			shutdown_ht_cleanup_thread,
+			cleanup_ht_cleanup_thread,
+			NULL);
+	if (!thread) {
+		goto error;
+	}
+	return thread;
+error:
+	cleanup_ht_cleanup_thread(NULL);
+	return NULL;
+}