[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread.c

/*
 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
 *
 * SPDX-License-Identifier: GPL-2.0-only
 *
 */

#define _LGPL_SOURCE
#include <lttng/trigger/trigger.h>
#include <lttng/notification/channel-internal.h>
#include <lttng/notification/notification-internal.h>
#include <lttng/condition/condition-internal.h>
#include <lttng/condition/buffer-usage-internal.h>
#include <common/error.h>
#include <common/config/session-config.h>
#include <common/defaults.h>
#include <common/utils.h>
#include <common/align.h>
#include <common/time.h>
#include <sys/stat.h>
#include <time.h>
#include <signal.h>

#include "notification-thread.h"
#include "notification-thread-events.h"
#include "notification-thread-commands.h"
#include "lttng-sessiond.h"
#include "health-sessiond.h"
#include "thread.h"

#include <urcu.h>
#include <urcu/list.h>
#include <urcu/rculfhash.h>

/*
 * Destroy the thread data previously created by the init function.
 */
void notification_thread_handle_destroy(
		struct notification_thread_handle *handle)
{
	int ret;

	if (!handle) {
		goto end;
	}

	assert(cds_list_empty(&handle->cmd_queue.list));
	pthread_mutex_destroy(&handle->cmd_queue.lock);
	sem_destroy(&handle->ready);

	if (handle->cmd_queue.event_pipe) {
		lttng_pipe_destroy(handle->cmd_queue.event_pipe);
	}
	if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
		ret = close(handle->channel_monitoring_pipes.ust32_consumer);
		if (ret) {
			PERROR("close 32-bit consumer channel monitoring pipe");
		}
	}
	if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
		ret = close(handle->channel_monitoring_pipes.ust64_consumer);
		if (ret) {
			PERROR("close 64-bit consumer channel monitoring pipe");
		}
	}
	if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
		ret = close(handle->channel_monitoring_pipes.kernel_consumer);
		if (ret) {
			PERROR("close kernel consumer channel monitoring pipe");
		}
	}
end:
	free(handle);
}

struct notification_thread_handle *notification_thread_handle_create(
		struct lttng_pipe *ust32_channel_monitor_pipe,
		struct lttng_pipe *ust64_channel_monitor_pipe,
		struct lttng_pipe *kernel_channel_monitor_pipe)
{
	int ret;
	struct notification_thread_handle *handle;
	struct lttng_pipe *event_pipe = NULL;

	handle = zmalloc(sizeof(*handle));
	if (!handle) {
		goto end;
	}

	sem_init(&handle->ready, 0, 0);

	event_pipe = lttng_pipe_open(FD_CLOEXEC);
	if (!event_pipe) {
		ERR("event_pipe creation");
		goto error;
	}

	handle->cmd_queue.event_pipe = event_pipe;
	event_pipe = NULL;

	CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
	ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
	if (ret) {
		goto error;
	}

	if (ust32_channel_monitor_pipe) {
		handle->channel_monitoring_pipes.ust32_consumer =
				lttng_pipe_release_readfd(
					ust32_channel_monitor_pipe);
		if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
			goto error;
		}
	} else {
		handle->channel_monitoring_pipes.ust32_consumer = -1;
	}
	if (ust64_channel_monitor_pipe) {
		handle->channel_monitoring_pipes.ust64_consumer =
				lttng_pipe_release_readfd(
					ust64_channel_monitor_pipe);
		if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
			goto error;
		}
	} else {
		handle->channel_monitoring_pipes.ust64_consumer = -1;
	}
	if (kernel_channel_monitor_pipe) {
		handle->channel_monitoring_pipes.kernel_consumer =
				lttng_pipe_release_readfd(
					kernel_channel_monitor_pipe);
		if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
			goto error;
		}
	} else {
		handle->channel_monitoring_pipes.kernel_consumer = -1;
	}

end:
	return handle;
error:
	lttng_pipe_destroy(event_pipe);
	notification_thread_handle_destroy(handle);
	return NULL;
}

static
char *get_notification_channel_sock_path(void)
{
	int ret;
	bool is_root = !getuid();
	char *sock_path;

	sock_path = zmalloc(LTTNG_PATH_MAX);
	if (!sock_path) {
		goto error;
	}

	if (is_root) {
		ret = snprintf(sock_path, LTTNG_PATH_MAX,
				DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
		if (ret < 0) {
			goto error;
		}
	} else {
		const char *home_path = utils_get_home_dir();

		if (!home_path) {
			ERR("Can't get HOME directory for socket creation");
			goto error;
		}

		ret = snprintf(sock_path, LTTNG_PATH_MAX,
				DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
				home_path);
		if (ret < 0) {
			goto error;
		}
	}

	return sock_path;
error:
	free(sock_path);
	return NULL;
}

static
void notification_channel_socket_destroy(int fd)
{
	int ret;
	char *sock_path = get_notification_channel_sock_path();

	DBG("[notification-thread] Destroying notification channel socket");

	if (sock_path) {
		ret = unlink(sock_path);
		free(sock_path);
		if (ret < 0) {
			PERROR("unlink notification channel socket");
		}
	}

	ret = close(fd);
	if (ret) {
		PERROR("close notification channel socket");
	}
}

static
int notification_channel_socket_create(void)
{
	int fd = -1, ret;
	char *sock_path = get_notification_channel_sock_path();

	DBG("[notification-thread] Creating notification channel UNIX socket at %s",
			sock_path);

	ret = lttcomm_create_unix_sock(sock_path);
	if (ret < 0) {
		ERR("[notification-thread] Failed to create notification socket");
		goto error;
	}
	fd = ret;

	ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
	if (ret < 0) {
		ERR("Set file permissions failed: %s", sock_path);
		PERROR("chmod notification channel socket");
		goto error;
	}

	if (getuid() == 0) {
		gid_t gid;

		ret =  utils_get_group_id(config.tracing_group_name.value, true,
				&gid);
		if (ret) {
			/* Default to root group. */
			gid = 0;
		}

		ret = chown(sock_path, 0, gid);
		if (ret) {
			ERR("Failed to set the notification channel socket's group");
			ret = -1;
			goto error;
		}
	}

	DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
			fd);
	free(sock_path);
	return fd;
error:
	if (fd >= 0 && close(fd) < 0) {
		PERROR("close notification channel socket");
	}
	free(sock_path);
	return ret;
}

static
int init_poll_set(struct lttng_poll_event *poll_set,
		struct notification_thread_handle *handle,
		int notification_channel_socket)
{
	int ret;

	/*
	 * Create pollset with size 5:
	 *	- notification channel socket (listen for new connections),
	 *	- command queue event fd (internal sessiond commands),
	 *	- consumerd (32-bit user space) channel monitor pipe,
	 *	- consumerd (64-bit user space) channel monitor pipe,
	 *	- consumerd (kernel) channel monitor pipe.
	 */
	ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
	if (ret < 0) {
		goto end;
	}

	ret = lttng_poll_add(poll_set, notification_channel_socket,
			LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
	if (ret < 0) {
		ERR("[notification-thread] Failed to add notification channel socket to pollset");
		goto error;
	}
	ret = lttng_poll_add(poll_set, lttng_pipe_get_readfd(handle->cmd_queue.event_pipe),
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
		goto error;
	}
	ret = lttng_poll_add(poll_set,
			handle->channel_monitoring_pipes.ust32_consumer,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
		goto error;
	}
	ret = lttng_poll_add(poll_set,
			handle->channel_monitoring_pipes.ust64_consumer,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
		goto error;
	}
	if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
		goto end;
	}
	ret = lttng_poll_add(poll_set,
			handle->channel_monitoring_pipes.kernel_consumer,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
		goto error;
	}
end:
	return ret;
error:
	lttng_poll_clean(poll_set);
	return ret;
}

static
void fini_thread_state(struct notification_thread_state *state)
{
	int ret;

	if (state->client_socket_ht) {
		ret = handle_notification_thread_client_disconnect_all(state);
		assert(!ret);
		ret = cds_lfht_destroy(state->client_socket_ht, NULL);
		assert(!ret);
	}
	if (state->client_id_ht) {
		ret = cds_lfht_destroy(state->client_id_ht, NULL);
		assert(!ret);
	}
	if (state->triggers_ht) {
		ret = handle_notification_thread_trigger_unregister_all(state);
		assert(!ret);
		ret = cds_lfht_destroy(state->triggers_ht, NULL);
		assert(!ret);
	}
	if (state->channel_triggers_ht) {
		ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
		assert(!ret);
	}
	if (state->channel_state_ht) {
		ret = cds_lfht_destroy(state->channel_state_ht, NULL);
		assert(!ret);
	}
	if (state->notification_trigger_clients_ht) {
		ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
				NULL);
		assert(!ret);
	}
	if (state->channels_ht) {
		ret = cds_lfht_destroy(state->channels_ht, NULL);
		assert(!ret);
	}
	if (state->sessions_ht) {
		ret = cds_lfht_destroy(state->sessions_ht, NULL);
		assert(!ret);
	}
	if (state->triggers_by_name_uid_ht) {
		ret = cds_lfht_destroy(state->triggers_by_name_uid_ht, NULL);
		assert(!ret);
	}
	if (state->trigger_tokens_ht) {
		ret = cds_lfht_destroy(state->trigger_tokens_ht, NULL);
		assert(!ret);
	}
	/*
	 * Must be destroyed after all channels have been destroyed.
	 * See comment in struct lttng_session_trigger_list.
	 */
	if (state->session_triggers_ht) {
		ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
		assert(!ret);
	}
	if (state->notification_channel_socket >= 0) {
		notification_channel_socket_destroy(
				state->notification_channel_socket);
	}

	assert(cds_list_empty(&state->tracer_event_sources_list));

	if (state->executor) {
		action_executor_destroy(state->executor);
	}
	lttng_poll_clean(&state->events);
}

static
void mark_thread_as_ready(struct notification_thread_handle *handle)
{
	DBG("Marking notification thread as ready");
	sem_post(&handle->ready);
}

static
void wait_until_thread_is_ready(struct notification_thread_handle *handle)
{
	DBG("Waiting for notification thread to be ready");
	sem_wait(&handle->ready);
	DBG("Notification thread is ready");
}

static
int init_thread_state(struct notification_thread_handle *handle,
		struct notification_thread_state *state)
{
	int ret;

	memset(state, 0, sizeof(*state));
	state->notification_channel_socket = -1;
	state->trigger_id.next_tracer_token = 1;
	lttng_poll_init(&state->events);

	ret = notification_channel_socket_create();
	if (ret < 0) {
		goto end;
	}
	state->notification_channel_socket = ret;

	ret = init_poll_set(&state->events, handle,
			state->notification_channel_socket);
	if (ret) {
		goto end;
	}

	DBG("[notification-thread] Listening on notification channel socket");
	ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
	if (ret < 0) {
		ERR("[notification-thread] Listen failed on notification channel socket");
		goto error;
	}

	state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
			CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->client_socket_ht) {
		goto error;
	}

	state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
			CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->client_id_ht) {
		goto error;
	}

	state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
			CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->channel_triggers_ht) {
		goto error;
	}

	state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
			CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->session_triggers_ht) {
		goto error;
	}

	state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
			CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->channel_state_ht) {
		goto error;
	}

	state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->notification_trigger_clients_ht) {
		goto error;
	}

	state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->channels_ht) {
		goto error;
	}
	state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->sessions_ht) {
		goto error;
	}
	state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->triggers_ht) {
		goto error;
	}
	state->triggers_by_name_uid_ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->triggers_by_name_uid_ht) {
		goto error;
	}

	state->trigger_tokens_ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!state->trigger_tokens_ht) {
		goto error;
	}

	CDS_INIT_LIST_HEAD(&state->tracer_event_sources_list);

	state->executor = action_executor_create(handle);
	if (!state->executor) {
		goto error;
	}
	mark_thread_as_ready(handle);
end:
	return 0;
error:
	fini_thread_state(state);
	return -1;
}

static
int handle_channel_monitoring_pipe(int fd, uint32_t revents,
		struct notification_thread_handle *handle,
		struct notification_thread_state *state)
{
	int ret = 0;
	enum lttng_domain_type domain;

	if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
			fd == handle->channel_monitoring_pipes.ust64_consumer) {
		domain = LTTNG_DOMAIN_UST;
	} else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
		domain = LTTNG_DOMAIN_KERNEL;
	} else {
		abort();
	}

	if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
		ret = lttng_poll_del(&state->events, fd);
		if (ret) {
			ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
		}
		goto end;
	}

	ret = handle_notification_thread_channel_sample(
			state, fd, domain);
	if (ret) {
		ERR("[notification-thread] Consumer sample handling error occurred");
		ret = -1;
		goto end;
	}
end:
	return ret;
}

/*
 * This thread services notification channel clients and commands received
 * from various lttng-sessiond components over a command queue.
 */
static
void *thread_notification(void *data)
{
	int ret;
	struct notification_thread_handle *handle = data;
	struct notification_thread_state state;

	DBG("[notification-thread] Started notification thread");

	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
	rcu_register_thread();
	rcu_thread_online();

	if (!handle) {
		ERR("[notification-thread] Invalid thread context provided");
		goto end;
	}

	health_code_update();

	ret = init_thread_state(handle, &state);
	if (ret) {
		goto end;
	}

	while (true) {
		int fd_count, i;

		health_poll_entry();
		DBG("[notification-thread] Entering poll wait");
		ret = lttng_poll_wait(&state.events, -1);
		DBG("[notification-thread] Poll wait returned (%i)", ret);
		health_poll_exit();
		if (ret < 0) {
			/*
			 * Restart interrupted system call.
			 */
			if (errno == EINTR) {
				continue;
			}
			ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
			goto error;
		}

		fd_count = ret;
		for (i = 0; i < fd_count; i++) {
			int fd = LTTNG_POLL_GETFD(&state.events, i);
			uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);

			DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);

			if (fd == state.notification_channel_socket) {
				if (revents & LPOLLIN) {
					ret = handle_notification_thread_client_connect(
							&state);
					if (ret < 0) {
						goto error;
					}
				} else if (revents &
						(LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
					ERR("[notification-thread] Notification socket poll error");
					goto error;
				} else {
					ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
					goto error;
				}
			} else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
				ret = handle_notification_thread_command(handle,
						&state);
				if (ret < 0) {
					DBG("[notification-thread] Error encountered while servicing command queue");
					goto error;
				} else if (ret > 0) {
					goto exit;
				}
			} else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
					fd == handle->channel_monitoring_pipes.ust64_consumer ||
					fd == handle->channel_monitoring_pipes.kernel_consumer) {
				ret = handle_channel_monitoring_pipe(fd,
						revents, handle, &state);
				if (ret) {
					goto error;
				}
			} else {
				/* Activity on a client's socket. */
				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
					/*
					 * It doesn't matter if a command was
					 * pending on the client socket at this
					 * point since it now has no way to
					 * receive the notifications to which
					 * it was subscribing or unsubscribing.
					 */
					ret = handle_notification_thread_client_disconnect(
							fd, &state);
					if (ret) {
						goto error;
					}
				} else {
					if (revents & LPOLLIN) {
						ret = handle_notification_thread_client_in(
							&state, fd);
						if (ret) {
							goto error;
						}
					}

					if (revents & LPOLLOUT) {
						ret = handle_notification_thread_client_out(
							&state, fd);
						if (ret) {
							goto error;
						}
					}
				}
			}
		}
	}
exit:
error:
	fini_thread_state(&state);
end:
	rcu_thread_offline();
	rcu_unregister_thread();
	health_unregister(health_sessiond);
	return NULL;
}

static
bool shutdown_notification_thread(void *thread_data)
{
	struct notification_thread_handle *handle = thread_data;

	notification_thread_command_quit(handle);
	return true;
}

struct lttng_thread *launch_notification_thread(
		struct notification_thread_handle *handle)
{
	struct lttng_thread *thread;

	thread = lttng_thread_create("Notification",
			thread_notification,
			shutdown_notification_thread,
			NULL,
			handle);
	if (!thread) {
		goto error;
	}

	/*
	 * Wait for the thread to be marked as "ready" before returning
	 * as other subsystems depend on the notification subsystem
	 * (e.g. rotation thread).
	 */
	wait_until_thread_is_ready(handle);
	return thread;
error:
	return NULL;
}
Commit	Line	Data
ab0ee2ca	1	/*
ab5be9fa	2	* Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
ab0ee2ca	3	*
ab5be9fa	4	* SPDX-License-Identifier: GPL-2.0-only
ab0ee2ca	5	*
ab0ee2ca JG	6	*/
	7
	8	#define _LGPL_SOURCE
	9	#include <lttng/trigger/trigger.h>
	10	#include <lttng/notification/channel-internal.h>
	11	#include <lttng/notification/notification-internal.h>
	12	#include <lttng/condition/condition-internal.h>
	13	#include <lttng/condition/buffer-usage-internal.h>
	14	#include <common/error.h>
	15	#include <common/config/session-config.h>
	16	#include <common/defaults.h>
	17	#include <common/utils.h>
ab0ee2ca JG	18	#include <common/align.h>
ab0ee2ca JG	19	#include <common/time.h>
ab0ee2ca JG	20	#include <sys/stat.h>
	21	#include <time.h>
	22	#include <signal.h>
	23
	24	#include "notification-thread.h"
	25	#include "notification-thread-events.h"
	26	#include "notification-thread-commands.h"
	27	#include "lttng-sessiond.h"
	28	#include "health-sessiond.h"
c8a9de5a	29	#include "thread.h"
ab0ee2ca JG	30
	31	#include <urcu.h>
	32	#include <urcu/list.h>
	33	#include <urcu/rculfhash.h>
	34
ab0ee2ca JG	35	/*
	36	* Destroy the thread data previously created by the init function.
	37	*/
	38	void notification_thread_handle_destroy(
	39	struct notification_thread_handle *handle)
	40	{
	41	int ret;
ab0ee2ca JG	42
	43	if (!handle) {
	44	goto end;
	45	}
	46
8ada111f	47	assert(cds_list_empty(&handle->cmd_queue.list));
ab0ee2ca	48	pthread_mutex_destroy(&handle->cmd_queue.lock);
c8a9de5a	49	sem_destroy(&handle->ready);
ab0ee2ca	50
814b4934 JR	51	if (handle->cmd_queue.event_pipe) {
	52	lttng_pipe_destroy(handle->cmd_queue.event_pipe);
	53	}
ab0ee2ca JG	54	if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
	55	ret = close(handle->channel_monitoring_pipes.ust32_consumer);
	56	if (ret) {
	57	PERROR("close 32-bit consumer channel monitoring pipe");
	58	}
	59	}
	60	if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
	61	ret = close(handle->channel_monitoring_pipes.ust64_consumer);
	62	if (ret) {
	63	PERROR("close 64-bit consumer channel monitoring pipe");
	64	}
	65	}
	66	if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
	67	ret = close(handle->channel_monitoring_pipes.kernel_consumer);
	68	if (ret) {
	69	PERROR("close kernel consumer channel monitoring pipe");
	70	}
	71	}
	72	end:
	73	free(handle);
	74	}
	75
	76	struct notification_thread_handle *notification_thread_handle_create(
	77	struct lttng_pipe *ust32_channel_monitor_pipe,
	78	struct lttng_pipe *ust64_channel_monitor_pipe,
c8a9de5a	79	struct lttng_pipe *kernel_channel_monitor_pipe)
ab0ee2ca JG	80	{
	81	int ret;
	82	struct notification_thread_handle *handle;
814b4934	83	struct lttng_pipe *event_pipe = NULL;
ab0ee2ca JG	84
	85	handle = zmalloc(sizeof(*handle));
	86	if (!handle) {
	87	goto end;
	88	}
	89
c8a9de5a JG	90	sem_init(&handle->ready, 0, 0);
c8a9de5a JG	91
18d08850	92	event_pipe = lttng_pipe_open(FD_CLOEXEC);
814b4934 JR	93	if (!event_pipe) {
814b4934 JR	94	ERR("event_pipe creation");
ab0ee2ca JG	95	goto error;
ab0ee2ca JG	96	}
814b4934 JR	97
	98	handle->cmd_queue.event_pipe = event_pipe;
	99	event_pipe = NULL;
	100
ab0ee2ca JG	101	CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
	102	ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
	103	if (ret) {
	104	goto error;
	105	}
	106
	107	if (ust32_channel_monitor_pipe) {
	108	handle->channel_monitoring_pipes.ust32_consumer =
	109	lttng_pipe_release_readfd(
	110	ust32_channel_monitor_pipe);
	111	if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
	112	goto error;
	113	}
	114	} else {
	115	handle->channel_monitoring_pipes.ust32_consumer = -1;
	116	}
	117	if (ust64_channel_monitor_pipe) {
	118	handle->channel_monitoring_pipes.ust64_consumer =
	119	lttng_pipe_release_readfd(
	120	ust64_channel_monitor_pipe);
	121	if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
	122	goto error;
	123	}
	124	} else {
	125	handle->channel_monitoring_pipes.ust64_consumer = -1;
	126	}
	127	if (kernel_channel_monitor_pipe) {
	128	handle->channel_monitoring_pipes.kernel_consumer =
	129	lttng_pipe_release_readfd(
	130	kernel_channel_monitor_pipe);
	131	if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
	132	goto error;
	133	}
	134	} else {
	135	handle->channel_monitoring_pipes.kernel_consumer = -1;
	136	}
d02d7404	137
ab0ee2ca JG	138	end:
	139	return handle;
	140	error:
814b4934	141	lttng_pipe_destroy(event_pipe);
ab0ee2ca JG	142	notification_thread_handle_destroy(handle);
	143	return NULL;
	144	}
	145
	146	static
	147	char *get_notification_channel_sock_path(void)
	148	{
	149	int ret;
	150	bool is_root = !getuid();
	151	char *sock_path;
	152
	153	sock_path = zmalloc(LTTNG_PATH_MAX);
	154	if (!sock_path) {
	155	goto error;
	156	}
	157
	158	if (is_root) {
	159	ret = snprintf(sock_path, LTTNG_PATH_MAX,
	160	DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
	161	if (ret < 0) {
	162	goto error;
	163	}
	164	} else {
4f00620d	165	const char *home_path = utils_get_home_dir();
ab0ee2ca JG	166
	167	if (!home_path) {
	168	ERR("Can't get HOME directory for socket creation");
	169	goto error;
	170	}
	171
	172	ret = snprintf(sock_path, LTTNG_PATH_MAX,
	173	DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
	174	home_path);
	175	if (ret < 0) {
	176	goto error;
	177	}
	178	}
	179
	180	return sock_path;
	181	error:
	182	free(sock_path);
	183	return NULL;
	184	}
	185
	186	static
	187	void notification_channel_socket_destroy(int fd)
	188	{
	189	int ret;
	190	char *sock_path = get_notification_channel_sock_path();
	191
	192	DBG("[notification-thread] Destroying notification channel socket");
	193
	194	if (sock_path) {
	195	ret = unlink(sock_path);
	196	free(sock_path);
	197	if (ret < 0) {
	198	PERROR("unlink notification channel socket");
	199	}
	200	}
	201
	202	ret = close(fd);
	203	if (ret) {
	204	PERROR("close notification channel socket");
	205	}
	206	}
	207
	208	static
	209	int notification_channel_socket_create(void)
	210	{
	211	int fd = -1, ret;
	212	char *sock_path = get_notification_channel_sock_path();
	213
	214	DBG("[notification-thread] Creating notification channel UNIX socket at %s",
	215	sock_path);
	216
	217	ret = lttcomm_create_unix_sock(sock_path);
	218	if (ret < 0) {
	219	ERR("[notification-thread] Failed to create notification socket");
	220	goto error;
	221	}
	222	fd = ret;
	223
	224	ret = chmod(sock_path, S_IRUSR \| S_IWUSR \| S_IRGRP \| S_IWGRP);
	225	if (ret < 0) {
	226	ERR("Set file permissions failed: %s", sock_path);
	227	PERROR("chmod notification channel socket");
	228	goto error;
	229	}
230
231	if (getuid() == 0) {
28ab59d0 JR	232	gid_t gid;
	233
	234	ret = utils_get_group_id(config.tracing_group_name.value, true,
	235	&gid);
	236	if (ret) {
	237	/* Default to root group. */
	238	gid = 0;
	239	}
	240
	241	ret = chown(sock_path, 0, gid);
ab0ee2ca JG	242	if (ret) {
	243	ERR("Failed to set the notification channel socket's group");
	244	ret = -1;
	245	goto error;
	246	}
	247	}
	248
	249	DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
	250	fd);
	251	free(sock_path);
	252	return fd;
	253	error:
	254	if (fd >= 0 && close(fd) < 0) {
	255	PERROR("close notification channel socket");
	256	}
	257	free(sock_path);
	258	return ret;
	259	}
	260
	261	static
	262	int init_poll_set(struct lttng_poll_event *poll_set,
	263	struct notification_thread_handle *handle,
	264	int notification_channel_socket)
	265	{
	266	int ret;
	267
	268	/*
	269	* Create pollset with size 5:
	270	* - notification channel socket (listen for new connections),
	271	* - command queue event fd (internal sessiond commands),
	272	* - consumerd (32-bit user space) channel monitor pipe,
	273	* - consumerd (64-bit user space) channel monitor pipe,
	274	* - consumerd (kernel) channel monitor pipe.
	275	*/
	276	ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
	277	if (ret < 0) {
	278	goto end;
	279	}
	280
	281	ret = lttng_poll_add(poll_set, notification_channel_socket,
	282	LPOLLIN \| LPOLLERR \| LPOLLHUP \| LPOLLRDHUP);
	283	if (ret < 0) {
	284	ERR("[notification-thread] Failed to add notification channel socket to pollset");
	285	goto error;
	286	}
814b4934	287	ret = lttng_poll_add(poll_set, lttng_pipe_get_readfd(handle->cmd_queue.event_pipe),
ab0ee2ca JG	288	LPOLLIN \| LPOLLERR);
	289	if (ret < 0) {
	290	ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
	291	goto error;
	292	}
	293	ret = lttng_poll_add(poll_set,
	294	handle->channel_monitoring_pipes.ust32_consumer,
	295	LPOLLIN \| LPOLLERR);
	296	if (ret < 0) {
	297	ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
	298	goto error;
	299	}
	300	ret = lttng_poll_add(poll_set,
	301	handle->channel_monitoring_pipes.ust64_consumer,
	302	LPOLLIN \| LPOLLERR);
	303	if (ret < 0) {
	304	ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
	305	goto error;
	306	}
	307	if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
	308	goto end;
	309	}
	310	ret = lttng_poll_add(poll_set,
	311	handle->channel_monitoring_pipes.kernel_consumer,
	312	LPOLLIN \| LPOLLERR);
	313	if (ret < 0) {
	314	ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
	315	goto error;
	316	}
	317	end:
	318	return ret;
	319	error:
	320	lttng_poll_clean(poll_set);
	321	return ret;
	322	}
	323
	324	static
	325	void fini_thread_state(struct notification_thread_state *state)
	326	{
	327	int ret;
	328
	329	if (state->client_socket_ht) {
	330	ret = handle_notification_thread_client_disconnect_all(state);
	331	assert(!ret);
	332	ret = cds_lfht_destroy(state->client_socket_ht, NULL);
	333	assert(!ret);
	334	}
ac1889bf JG	335	if (state->client_id_ht) {
	336	ret = cds_lfht_destroy(state->client_id_ht, NULL);
	337	assert(!ret);
	338	}
ab0ee2ca JG	339	if (state->triggers_ht) {
	340	ret = handle_notification_thread_trigger_unregister_all(state);
	341	assert(!ret);
	342	ret = cds_lfht_destroy(state->triggers_ht, NULL);
	343	assert(!ret);
	344	}
	345	if (state->channel_triggers_ht) {
	346	ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
	347	assert(!ret);
	348	}
	349	if (state->channel_state_ht) {
	350	ret = cds_lfht_destroy(state->channel_state_ht, NULL);
	351	assert(!ret);
	352	}
	353	if (state->notification_trigger_clients_ht) {
	354	ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
	355	NULL);
	356	assert(!ret);
	357	}
	358	if (state->channels_ht) {
8abe313a JG	359	ret = cds_lfht_destroy(state->channels_ht, NULL);
	360	assert(!ret);
	361	}
	362	if (state->sessions_ht) {
	363	ret = cds_lfht_destroy(state->sessions_ht, NULL);
ab0ee2ca JG	364	assert(!ret);
ab0ee2ca JG	365	}
242388e4 JR	366	if (state->triggers_by_name_uid_ht) {
	367	ret = cds_lfht_destroy(state->triggers_by_name_uid_ht, NULL);
	368	assert(!ret);
	369	}
e7c93cf9 JR	370	if (state->trigger_tokens_ht) {
	371	ret = cds_lfht_destroy(state->trigger_tokens_ht, NULL);
	372	assert(!ret);
	373	}
ea9a44f0 JG	374	/*
	375	* Must be destroyed after all channels have been destroyed.
	376	* See comment in struct lttng_session_trigger_list.
	377	*/
	378	if (state->session_triggers_ht) {
	379	ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
	380	assert(!ret);
	381	}
ab0ee2ca JG	382	if (state->notification_channel_socket >= 0) {
	383	notification_channel_socket_destroy(
	384	state->notification_channel_socket);
	385	}
d02d7404 JR	386
	387	assert(cds_list_empty(&state->tracer_event_sources_list));
	388
f2b3ef9f JG	389	if (state->executor) {
	390	action_executor_destroy(state->executor);
	391	}
ab0ee2ca JG	392	lttng_poll_clean(&state->events);
	393	}
	394
c8a9de5a JG	395	static
	396	void mark_thread_as_ready(struct notification_thread_handle *handle)
	397	{
	398	DBG("Marking notification thread as ready");
	399	sem_post(&handle->ready);
	400	}
	401
	402	static
	403	void wait_until_thread_is_ready(struct notification_thread_handle *handle)
	404	{
	405	DBG("Waiting for notification thread to be ready");
	406	sem_wait(&handle->ready);
	407	DBG("Notification thread is ready");
	408	}
	409
ab0ee2ca JG	410	static
	411	int init_thread_state(struct notification_thread_handle *handle,
	412	struct notification_thread_state *state)
	413	{
	414	int ret;
	415
	416	memset(state, 0, sizeof(*state));
	417	state->notification_channel_socket = -1;
e6887944	418	state->trigger_id.next_tracer_token = 1;
ab0ee2ca JG	419	lttng_poll_init(&state->events);
	420
	421	ret = notification_channel_socket_create();
	422	if (ret < 0) {
	423	goto end;
	424	}
	425	state->notification_channel_socket = ret;
	426
	427	ret = init_poll_set(&state->events, handle,
	428	state->notification_channel_socket);
	429	if (ret) {
	430	goto end;
	431	}
	432
	433	DBG("[notification-thread] Listening on notification channel socket");
	434	ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
	435	if (ret < 0) {
	436	ERR("[notification-thread] Listen failed on notification channel socket");
	437	goto error;
	438	}
	439
	440	state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	441	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	442	if (!state->client_socket_ht) {
	443	goto error;
	444	}
	445
ac1889bf JG	446	state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	447	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	448	if (!state->client_id_ht) {
	449	goto error;
	450	}
	451
ab0ee2ca JG	452	state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	453	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	454	if (!state->channel_triggers_ht) {
	455	goto error;
	456	}
	457
ea9a44f0 JG	458	state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	459	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	460	if (!state->session_triggers_ht) {
	461	goto error;
	462	}
	463
ab0ee2ca JG	464	state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	465	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	466	if (!state->channel_state_ht) {
	467	goto error;
	468	}
	469
	470	state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	471	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	472	if (!state->notification_trigger_clients_ht) {
	473	goto error;
	474	}
	475
	476	state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	477	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	478	if (!state->channels_ht) {
	479	goto error;
	480	}
8abe313a JG	481	state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	482	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	483	if (!state->sessions_ht) {
	484	goto error;
	485	}
ab0ee2ca JG	486	state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	487	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	488	if (!state->triggers_ht) {
	489	goto error;
f2b3ef9f	490	}
242388e4 JR	491	state->triggers_by_name_uid_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	492	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	493	if (!state->triggers_by_name_uid_ht) {
	494	goto error;
	495	}
f2b3ef9f	496
e7c93cf9 JR	497	state->trigger_tokens_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	498	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	499	if (!state->trigger_tokens_ht) {
	500	goto error;
	501	}
	502
d02d7404 JR	503	CDS_INIT_LIST_HEAD(&state->tracer_event_sources_list);
d02d7404 JR	504
f2b3ef9f JG	505	state->executor = action_executor_create(handle);
	506	if (!state->executor) {
	507	goto error;
ab0ee2ca	508	}
c8a9de5a	509	mark_thread_as_ready(handle);
ab0ee2ca JG	510	end:
	511	return 0;
	512	error:
	513	fini_thread_state(state);
	514	return -1;
	515	}
	516
	517	static
	518	int handle_channel_monitoring_pipe(int fd, uint32_t revents,
	519	struct notification_thread_handle *handle,
	520	struct notification_thread_state *state)
	521	{
	522	int ret = 0;
	523	enum lttng_domain_type domain;
	524
	525	if (fd == handle->channel_monitoring_pipes.ust32_consumer \|\|
	526	fd == handle->channel_monitoring_pipes.ust64_consumer) {
	527	domain = LTTNG_DOMAIN_UST;
	528	} else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
	529	domain = LTTNG_DOMAIN_KERNEL;
	530	} else {
	531	abort();
	532	}
	533
	534	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	535	ret = lttng_poll_del(&state->events, fd);
	536	if (ret) {
	537	ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
	538	}
	539	goto end;
	540	}
	541
	542	ret = handle_notification_thread_channel_sample(
	543	state, fd, domain);
	544	if (ret) {
4149ace8	545	ERR("[notification-thread] Consumer sample handling error occurred");
ab0ee2ca JG	546	ret = -1;
	547	goto end;
	548	}
	549	end:
	550	return ret;
	551	}
	552
	553	/*
	554	* This thread services notification channel clients and commands received
	555	* from various lttng-sessiond components over a command queue.
	556	*/
c8a9de5a	557	static
ab0ee2ca JG	558	void thread_notification(void data)
	559	{
	560	int ret;
	561	struct notification_thread_handle *handle = data;
	562	struct notification_thread_state state;
	563
	564	DBG("[notification-thread] Started notification thread");
	565
f620cc28 JG	566	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
	567	rcu_register_thread();
	568	rcu_thread_online();
	569
ab0ee2ca JG	570	if (!handle) {
	571	ERR("[notification-thread] Invalid thread context provided");
	572	goto end;
	573	}
	574
ab0ee2ca JG	575	health_code_update();
	576
	577	ret = init_thread_state(handle, &state);
	578	if (ret) {
	579	goto end;
	580	}
	581
ab0ee2ca JG	582	while (true) {
	583	int fd_count, i;
	584
	585	health_poll_entry();
	586	DBG("[notification-thread] Entering poll wait");
	587	ret = lttng_poll_wait(&state.events, -1);
	588	DBG("[notification-thread] Poll wait returned (%i)", ret);
	589	health_poll_exit();
	590	if (ret < 0) {
	591	/*
	592	* Restart interrupted system call.
	593	*/
	594	if (errno == EINTR) {
	595	continue;
	596	}
	597	ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
	598	goto error;
	599	}
	600
	601	fd_count = ret;
	602	for (i = 0; i < fd_count; i++) {
	603	int fd = LTTNG_POLL_GETFD(&state.events, i);
	604	uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
	605
	606	DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
	607
	608	if (fd == state.notification_channel_socket) {
	609	if (revents & LPOLLIN) {
	610	ret = handle_notification_thread_client_connect(
	611	&state);
	612	if (ret < 0) {
	613	goto error;
	614	}
	615	} else if (revents &
	616	(LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	617	ERR("[notification-thread] Notification socket poll error");
	618	goto error;
	619	} else {
	620	ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
	621	goto error;
	622	}
814b4934	623	} else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
ab0ee2ca JG	624	ret = handle_notification_thread_command(handle,
	625	&state);
	626	if (ret < 0) {
	627	DBG("[notification-thread] Error encountered while servicing command queue");
	628	goto error;
	629	} else if (ret > 0) {
	630	goto exit;
	631	}
	632	} else if (fd == handle->channel_monitoring_pipes.ust32_consumer \|\|
	633	fd == handle->channel_monitoring_pipes.ust64_consumer \|\|
	634	fd == handle->channel_monitoring_pipes.kernel_consumer) {
	635	ret = handle_channel_monitoring_pipe(fd,
	636	revents, handle, &state);
	637	if (ret) {
	638	goto error;
	639	}
	640	} else {
	641	/* Activity on a client's socket. */
	642	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	643	/*
	644	* It doesn't matter if a command was
	645	* pending on the client socket at this
	646	* point since it now has no way to
	647	* receive the notifications to which
	648	* it was subscribing or unsubscribing.
	649	*/
	650	ret = handle_notification_thread_client_disconnect(
	651	fd, &state);
	652	if (ret) {
	653	goto error;
	654	}
	655	} else {
	656	if (revents & LPOLLIN) {
	657	ret = handle_notification_thread_client_in(
	658	&state, fd);
	659	if (ret) {
	660	goto error;
	661	}
	662	}
	663
	664	if (revents & LPOLLOUT) {
	665	ret = handle_notification_thread_client_out(
	666	&state, fd);
	667	if (ret) {
	668	goto error;
	669	}
	670	}
	671	}
	672	}
	673	}
	674	}
	675	exit:
	676	error:
	677	fini_thread_state(&state);
f620cc28	678	end:
ab0ee2ca JG	679	rcu_thread_offline();
ab0ee2ca JG	680	rcu_unregister_thread();
f620cc28	681	health_unregister(health_sessiond);
ab0ee2ca JG	682	return NULL;
ab0ee2ca JG	683	}
c8a9de5a JG	684
	685	static
	686	bool shutdown_notification_thread(void *thread_data)
	687	{
	688	struct notification_thread_handle *handle = thread_data;
	689
	690	notification_thread_command_quit(handle);
	691	return true;
	692	}
	693
4a91420c JG	694	struct lttng_thread *launch_notification_thread(
4a91420c JG	695	struct notification_thread_handle *handle)
c8a9de5a JG	696	{
	697	struct lttng_thread *thread;
	698
	699	thread = lttng_thread_create("Notification",
	700	thread_notification,
	701	shutdown_notification_thread,
	702	NULL,
	703	handle);
	704	if (!thread) {
	705	goto error;
	706	}
	707
	708	/*
	709	* Wait for the thread to be marked as "ready" before returning
	710	* as other subsystems depend on the notification subsystem
	711	* (e.g. rotation thread).
	712	*/
	713	wait_until_thread_is_ready(handle);
4a91420c	714	return thread;
c8a9de5a	715	error:
4a91420c	716	return NULL;
c8a9de5a	717	}