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