[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;
	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;
	410	lttng_poll_init(&state->events);
	411
	412	ret = notification_channel_socket_create();
	413	if (ret < 0) {
	414	goto end;
	415	}
	416	state->notification_channel_socket = ret;
	417
	418	ret = init_poll_set(&state->events, handle,
	419	state->notification_channel_socket);
	420	if (ret) {
	421	goto end;
	422	}
	423
	424	DBG("[notification-thread] Listening on notification channel socket");
	425	ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
	426	if (ret < 0) {
	427	ERR("[notification-thread] Listen failed on notification channel socket");
	428	goto error;
	429	}
	430
	431	state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	432	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	433	if (!state->client_socket_ht) {
	434	goto error;
	435	}
	436
ac1889bf JG	437	state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	438	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	439	if (!state->client_id_ht) {
	440	goto error;
	441	}
	442
ab0ee2ca JG	443	state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	444	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	445	if (!state->channel_triggers_ht) {
	446	goto error;
	447	}
	448
ea9a44f0 JG	449	state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	450	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	451	if (!state->session_triggers_ht) {
	452	goto error;
	453	}
	454
ab0ee2ca JG	455	state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	456	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	457	if (!state->channel_state_ht) {
	458	goto error;
	459	}
	460
	461	state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	462	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	463	if (!state->notification_trigger_clients_ht) {
	464	goto error;
	465	}
	466
	467	state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	468	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	469	if (!state->channels_ht) {
	470	goto error;
	471	}
8abe313a JG	472	state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	473	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	474	if (!state->sessions_ht) {
	475	goto error;
	476	}
ab0ee2ca JG	477	state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	478	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	479	if (!state->triggers_ht) {
	480	goto error;
f2b3ef9f	481	}
242388e4 JR	482	state->triggers_by_name_uid_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	483	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	484	if (!state->triggers_by_name_uid_ht) {
	485	goto error;
	486	}
f2b3ef9f JG	487
	488	state->executor = action_executor_create(handle);
	489	if (!state->executor) {
	490	goto error;
ab0ee2ca	491	}
c8a9de5a	492	mark_thread_as_ready(handle);
ab0ee2ca JG	493	end:
	494	return 0;
	495	error:
	496	fini_thread_state(state);
	497	return -1;
	498	}
	499
	500	static
	501	int handle_channel_monitoring_pipe(int fd, uint32_t revents,
	502	struct notification_thread_handle *handle,
	503	struct notification_thread_state *state)
	504	{
	505	int ret = 0;
	506	enum lttng_domain_type domain;
	507
	508	if (fd == handle->channel_monitoring_pipes.ust32_consumer \|\|
	509	fd == handle->channel_monitoring_pipes.ust64_consumer) {
	510	domain = LTTNG_DOMAIN_UST;
	511	} else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
	512	domain = LTTNG_DOMAIN_KERNEL;
	513	} else {
	514	abort();
	515	}
	516
	517	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	518	ret = lttng_poll_del(&state->events, fd);
	519	if (ret) {
	520	ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
	521	}
	522	goto end;
	523	}
	524
	525	ret = handle_notification_thread_channel_sample(
	526	state, fd, domain);
	527	if (ret) {
4149ace8	528	ERR("[notification-thread] Consumer sample handling error occurred");
ab0ee2ca JG	529	ret = -1;
	530	goto end;
	531	}
	532	end:
	533	return ret;
	534	}
	535
	536	/*
	537	* This thread services notification channel clients and commands received
	538	* from various lttng-sessiond components over a command queue.
	539	*/
c8a9de5a	540	static
ab0ee2ca JG	541	void thread_notification(void data)
	542	{
	543	int ret;
	544	struct notification_thread_handle *handle = data;
	545	struct notification_thread_state state;
	546
	547	DBG("[notification-thread] Started notification thread");
	548
f620cc28 JG	549	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
	550	rcu_register_thread();
	551	rcu_thread_online();
	552
ab0ee2ca JG	553	if (!handle) {
	554	ERR("[notification-thread] Invalid thread context provided");
	555	goto end;
	556	}
	557
ab0ee2ca JG	558	health_code_update();
	559
	560	ret = init_thread_state(handle, &state);
	561	if (ret) {
	562	goto end;
	563	}
	564
ab0ee2ca JG	565	while (true) {
	566	int fd_count, i;
	567
	568	health_poll_entry();
	569	DBG("[notification-thread] Entering poll wait");
	570	ret = lttng_poll_wait(&state.events, -1);
	571	DBG("[notification-thread] Poll wait returned (%i)", ret);
	572	health_poll_exit();
	573	if (ret < 0) {
	574	/*
	575	* Restart interrupted system call.
	576	*/
	577	if (errno == EINTR) {
	578	continue;
	579	}
	580	ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
	581	goto error;
	582	}
	583
	584	fd_count = ret;
	585	for (i = 0; i < fd_count; i++) {
	586	int fd = LTTNG_POLL_GETFD(&state.events, i);
	587	uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
	588
	589	DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
	590
	591	if (fd == state.notification_channel_socket) {
	592	if (revents & LPOLLIN) {
	593	ret = handle_notification_thread_client_connect(
	594	&state);
	595	if (ret < 0) {
	596	goto error;
	597	}
	598	} else if (revents &
	599	(LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	600	ERR("[notification-thread] Notification socket poll error");
	601	goto error;
	602	} else {
	603	ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
	604	goto error;
	605	}
814b4934	606	} else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
ab0ee2ca JG	607	ret = handle_notification_thread_command(handle,
	608	&state);
	609	if (ret < 0) {
	610	DBG("[notification-thread] Error encountered while servicing command queue");
	611	goto error;
	612	} else if (ret > 0) {
	613	goto exit;
	614	}
	615	} else if (fd == handle->channel_monitoring_pipes.ust32_consumer \|\|
	616	fd == handle->channel_monitoring_pipes.ust64_consumer \|\|
	617	fd == handle->channel_monitoring_pipes.kernel_consumer) {
	618	ret = handle_channel_monitoring_pipe(fd,
	619	revents, handle, &state);
	620	if (ret) {
	621	goto error;
	622	}
	623	} else {
	624	/* Activity on a client's socket. */
	625	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	626	/*
	627	* It doesn't matter if a command was
	628	* pending on the client socket at this
	629	* point since it now has no way to
	630	* receive the notifications to which
	631	* it was subscribing or unsubscribing.
	632	*/
	633	ret = handle_notification_thread_client_disconnect(
	634	fd, &state);
	635	if (ret) {
	636	goto error;
	637	}
	638	} else {
	639	if (revents & LPOLLIN) {
	640	ret = handle_notification_thread_client_in(
	641	&state, fd);
	642	if (ret) {
	643	goto error;
	644	}
	645	}
	646
	647	if (revents & LPOLLOUT) {
	648	ret = handle_notification_thread_client_out(
	649	&state, fd);
	650	if (ret) {
	651	goto error;
	652	}
	653	}
	654	}
	655	}
	656	}
	657	}
	658	exit:
	659	error:
	660	fini_thread_state(&state);
f620cc28	661	end:
ab0ee2ca JG	662	rcu_thread_offline();
ab0ee2ca JG	663	rcu_unregister_thread();
f620cc28	664	health_unregister(health_sessiond);
ab0ee2ca JG	665	return NULL;
ab0ee2ca JG	666	}
c8a9de5a JG	667
	668	static
	669	bool shutdown_notification_thread(void *thread_data)
	670	{
	671	struct notification_thread_handle *handle = thread_data;
	672
	673	notification_thread_command_quit(handle);
	674	return true;
	675	}
	676
4a91420c JG	677	struct lttng_thread *launch_notification_thread(
4a91420c JG	678	struct notification_thread_handle *handle)
c8a9de5a JG	679	{
	680	struct lttng_thread *thread;
	681
	682	thread = lttng_thread_create("Notification",
	683	thread_notification,
	684	shutdown_notification_thread,
	685	NULL,
	686	handle);
	687	if (!thread) {
	688	goto error;
	689	}
	690
	691	/*
	692	* Wait for the thread to be marked as "ready" before returning
	693	* as other subsystems depend on the notification subsystem
	694	* (e.g. rotation thread).
	695	*/
	696	wait_until_thread_is_ready(handle);
4a91420c	697	return thread;
c8a9de5a	698	error:
4a91420c	699	return NULL;
c8a9de5a	700	}