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

/*
 * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License, version 2 only, as
 * published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#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/eventfd.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 <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);

	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,
		sem_t *notification_thread_ready)
{
	int ret;
	struct notification_thread_handle *handle;
	struct lttng_pipe *event_pipe = NULL;

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

	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;
	}
	handle->notification_thread_ready = notification_thread_ready;
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 {
		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) {
		ret = chown(sock_path, 0,
				utils_get_group_id(config.tracing_group_name.value));
		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->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->notification_channel_socket >= 0) {
		notification_channel_socket_destroy(
				state->notification_channel_socket);
	}
	lttng_poll_clean(&state->events);
}

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->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->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;
	}
	sem_post(handle->notification_thread_ready);
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.
 */
void *thread_notification(void *data)
{
	int ret;
	struct notification_thread_handle *handle = data;
	struct notification_thread_state state;

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

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

	rcu_register_thread();
	rcu_thread_online();

	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
	health_code_update();

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

	/* Ready to handle client connections. */
	sessiond_notify_ready();

	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);

			if (!revents) {
				continue;
			}
			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);
	health_unregister(health_sessiond);
	rcu_thread_offline();
	rcu_unregister_thread();
end:
	return NULL;
}
Commit	Line	Data
ab0ee2ca JG	1	/*
	2	* Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
	3	*
	4	* This program is free software; you can redistribute it and/or modify it
	5	* under the terms of the GNU General Public License, version 2 only, as
	6	* published by the Free Software Foundation.
	7	*
	8	* This program is distributed in the hope that it will be useful, but WITHOUT
	9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	10	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	11	* more details.
	12	*
	13	* You should have received a copy of the GNU General Public License along with
	14	* this program; if not, write to the Free Software Foundation, Inc., 51
	15	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	16	*/
	17
	18	#define _LGPL_SOURCE
	19	#include <lttng/trigger/trigger.h>
	20	#include <lttng/notification/channel-internal.h>
	21	#include <lttng/notification/notification-internal.h>
	22	#include <lttng/condition/condition-internal.h>
	23	#include <lttng/condition/buffer-usage-internal.h>
	24	#include <common/error.h>
	25	#include <common/config/session-config.h>
	26	#include <common/defaults.h>
	27	#include <common/utils.h>
ab0ee2ca JG	28	#include <common/align.h>
	29	#include <common/time.h>
	30	#include <sys/eventfd.h>
	31	#include <sys/stat.h>
	32	#include <time.h>
	33	#include <signal.h>
	34
	35	#include "notification-thread.h"
	36	#include "notification-thread-events.h"
	37	#include "notification-thread-commands.h"
	38	#include "lttng-sessiond.h"
	39	#include "health-sessiond.h"
	40
	41	#include <urcu.h>
	42	#include <urcu/list.h>
	43	#include <urcu/rculfhash.h>
	44
ab0ee2ca JG	45	/*
	46	* Destroy the thread data previously created by the init function.
	47	*/
	48	void notification_thread_handle_destroy(
	49	struct notification_thread_handle *handle)
	50	{
	51	int ret;
ab0ee2ca JG	52
	53	if (!handle) {
	54	goto end;
	55	}
	56
8ada111f	57	assert(cds_list_empty(&handle->cmd_queue.list));
ab0ee2ca JG	58	pthread_mutex_destroy(&handle->cmd_queue.lock);
ab0ee2ca JG	59
814b4934 JR	60	if (handle->cmd_queue.event_pipe) {
	61	lttng_pipe_destroy(handle->cmd_queue.event_pipe);
	62	}
ab0ee2ca JG	63	if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
	64	ret = close(handle->channel_monitoring_pipes.ust32_consumer);
	65	if (ret) {
	66	PERROR("close 32-bit consumer channel monitoring pipe");
	67	}
	68	}
	69	if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
	70	ret = close(handle->channel_monitoring_pipes.ust64_consumer);
	71	if (ret) {
	72	PERROR("close 64-bit consumer channel monitoring pipe");
	73	}
	74	}
	75	if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
	76	ret = close(handle->channel_monitoring_pipes.kernel_consumer);
	77	if (ret) {
	78	PERROR("close kernel consumer channel monitoring pipe");
	79	}
	80	}
	81	end:
	82	free(handle);
	83	}
	84
	85	struct notification_thread_handle *notification_thread_handle_create(
	86	struct lttng_pipe *ust32_channel_monitor_pipe,
	87	struct lttng_pipe *ust64_channel_monitor_pipe,
90936dcf JD	88	struct lttng_pipe *kernel_channel_monitor_pipe,
90936dcf JD	89	sem_t *notification_thread_ready)
ab0ee2ca JG	90	{
	91	int ret;
	92	struct notification_thread_handle *handle;
814b4934	93	struct lttng_pipe *event_pipe = NULL;
ab0ee2ca JG	94
	95	handle = zmalloc(sizeof(*handle));
	96	if (!handle) {
	97	goto end;
	98	}
	99
18d08850	100	event_pipe = lttng_pipe_open(FD_CLOEXEC);
814b4934 JR	101	if (!event_pipe) {
814b4934 JR	102	ERR("event_pipe creation");
ab0ee2ca JG	103	goto error;
ab0ee2ca JG	104	}
814b4934 JR	105
	106	handle->cmd_queue.event_pipe = event_pipe;
	107	event_pipe = NULL;
	108
ab0ee2ca JG	109	CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
	110	ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
	111	if (ret) {
	112	goto error;
	113	}
	114
	115	if (ust32_channel_monitor_pipe) {
	116	handle->channel_monitoring_pipes.ust32_consumer =
	117	lttng_pipe_release_readfd(
	118	ust32_channel_monitor_pipe);
	119	if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
	120	goto error;
	121	}
	122	} else {
	123	handle->channel_monitoring_pipes.ust32_consumer = -1;
	124	}
	125	if (ust64_channel_monitor_pipe) {
	126	handle->channel_monitoring_pipes.ust64_consumer =
	127	lttng_pipe_release_readfd(
	128	ust64_channel_monitor_pipe);
	129	if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
	130	goto error;
	131	}
	132	} else {
	133	handle->channel_monitoring_pipes.ust64_consumer = -1;
	134	}
	135	if (kernel_channel_monitor_pipe) {
	136	handle->channel_monitoring_pipes.kernel_consumer =
	137	lttng_pipe_release_readfd(
	138	kernel_channel_monitor_pipe);
	139	if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
	140	goto error;
	141	}
	142	} else {
	143	handle->channel_monitoring_pipes.kernel_consumer = -1;
	144	}
90936dcf	145	handle->notification_thread_ready = notification_thread_ready;
ab0ee2ca JG	146	end:
	147	return handle;
	148	error:
814b4934	149	lttng_pipe_destroy(event_pipe);
ab0ee2ca JG	150	notification_thread_handle_destroy(handle);
	151	return NULL;
	152	}
	153
	154	static
	155	char *get_notification_channel_sock_path(void)
	156	{
	157	int ret;
	158	bool is_root = !getuid();
	159	char *sock_path;
	160
	161	sock_path = zmalloc(LTTNG_PATH_MAX);
	162	if (!sock_path) {
	163	goto error;
	164	}
	165
	166	if (is_root) {
	167	ret = snprintf(sock_path, LTTNG_PATH_MAX,
	168	DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
	169	if (ret < 0) {
	170	goto error;
	171	}
	172	} else {
	173	char *home_path = utils_get_home_dir();
	174
	175	if (!home_path) {
	176	ERR("Can't get HOME directory for socket creation");
	177	goto error;
	178	}
	179
	180	ret = snprintf(sock_path, LTTNG_PATH_MAX,
	181	DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
	182	home_path);
	183	if (ret < 0) {
	184	goto error;
	185	}
	186	}
	187
	188	return sock_path;
	189	error:
	190	free(sock_path);
	191	return NULL;
	192	}
	193
	194	static
	195	void notification_channel_socket_destroy(int fd)
	196	{
	197	int ret;
	198	char *sock_path = get_notification_channel_sock_path();
	199
	200	DBG("[notification-thread] Destroying notification channel socket");
	201
	202	if (sock_path) {
	203	ret = unlink(sock_path);
	204	free(sock_path);
	205	if (ret < 0) {
	206	PERROR("unlink notification channel socket");
	207	}
	208	}
	209
	210	ret = close(fd);
	211	if (ret) {
	212	PERROR("close notification channel socket");
	213	}
214	}
215
216	static
217	int notification_channel_socket_create(void)
218	{
219	int fd = -1, ret;
220	char *sock_path = get_notification_channel_sock_path();
221
222	DBG("[notification-thread] Creating notification channel UNIX socket at %s",
223	sock_path);
224
225	ret = lttcomm_create_unix_sock(sock_path);
226	if (ret < 0) {
227	ERR("[notification-thread] Failed to create notification socket");
228	goto error;
229	}
230	fd = ret;
231
232	ret = chmod(sock_path, S_IRUSR \| S_IWUSR \| S_IRGRP \| S_IWGRP);
233	if (ret < 0) {
234	ERR("Set file permissions failed: %s", sock_path);
235	PERROR("chmod notification channel socket");
236	goto error;
237	}
238
239	if (getuid() == 0) {
240	ret = chown(sock_path, 0,
e6142f2e	241	utils_get_group_id(config.tracing_group_name.value));
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	}
	335	if (state->triggers_ht) {
	336	ret = handle_notification_thread_trigger_unregister_all(state);
	337	assert(!ret);
	338	ret = cds_lfht_destroy(state->triggers_ht, NULL);
	339	assert(!ret);
	340	}
	341	if (state->channel_triggers_ht) {
	342	ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
	343	assert(!ret);
	344	}
	345	if (state->channel_state_ht) {
	346	ret = cds_lfht_destroy(state->channel_state_ht, NULL);
	347	assert(!ret);
	348	}
	349	if (state->notification_trigger_clients_ht) {
	350	ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
	351	NULL);
352	assert(!ret);
353	}
354	if (state->channels_ht) {
8abe313a JG	355	ret = cds_lfht_destroy(state->channels_ht, NULL);
	356	assert(!ret);
	357	}
	358	if (state->sessions_ht) {
	359	ret = cds_lfht_destroy(state->sessions_ht, NULL);
ab0ee2ca JG	360	assert(!ret);
ab0ee2ca JG	361	}
ab0ee2ca JG	362	if (state->notification_channel_socket >= 0) {
	363	notification_channel_socket_destroy(
	364	state->notification_channel_socket);
	365	}
	366	lttng_poll_clean(&state->events);
	367	}
	368
	369	static
	370	int init_thread_state(struct notification_thread_handle *handle,
	371	struct notification_thread_state *state)
	372	{
	373	int ret;
	374
	375	memset(state, 0, sizeof(*state));
	376	state->notification_channel_socket = -1;
	377	lttng_poll_init(&state->events);
	378
	379	ret = notification_channel_socket_create();
	380	if (ret < 0) {
	381	goto end;
	382	}
	383	state->notification_channel_socket = ret;
	384
	385	ret = init_poll_set(&state->events, handle,
	386	state->notification_channel_socket);
	387	if (ret) {
	388	goto end;
	389	}
	390
	391	DBG("[notification-thread] Listening on notification channel socket");
	392	ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
	393	if (ret < 0) {
	394	ERR("[notification-thread] Listen failed on notification channel socket");
	395	goto error;
	396	}
	397
	398	state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	399	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	400	if (!state->client_socket_ht) {
	401	goto error;
	402	}
	403
	404	state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	405	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	406	if (!state->channel_triggers_ht) {
	407	goto error;
	408	}
	409
	410	state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
	411	CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	412	if (!state->channel_state_ht) {
	413	goto error;
	414	}
	415
	416	state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	417	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	418	if (!state->notification_trigger_clients_ht) {
	419	goto error;
	420	}
	421
	422	state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	423	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	424	if (!state->channels_ht) {
	425	goto error;
426	}
8abe313a JG	427	state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	428	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	429	if (!state->sessions_ht) {
	430	goto error;
	431	}
ab0ee2ca JG	432	state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	433	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	434	if (!state->triggers_ht) {
	435	goto error;
	436	}
90936dcf	437	sem_post(handle->notification_thread_ready);
ab0ee2ca JG	438	end:
	439	return 0;
	440	error:
	441	fini_thread_state(state);
	442	return -1;
	443	}
	444
	445	static
	446	int handle_channel_monitoring_pipe(int fd, uint32_t revents,
	447	struct notification_thread_handle *handle,
	448	struct notification_thread_state *state)
	449	{
	450	int ret = 0;
	451	enum lttng_domain_type domain;
	452
	453	if (fd == handle->channel_monitoring_pipes.ust32_consumer \|\|
	454	fd == handle->channel_monitoring_pipes.ust64_consumer) {
	455	domain = LTTNG_DOMAIN_UST;
	456	} else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
	457	domain = LTTNG_DOMAIN_KERNEL;
	458	} else {
	459	abort();
	460	}
	461
	462	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	463	ret = lttng_poll_del(&state->events, fd);
	464	if (ret) {
	465	ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
	466	}
	467	goto end;
	468	}
	469
	470	ret = handle_notification_thread_channel_sample(
	471	state, fd, domain);
	472	if (ret) {
4149ace8	473	ERR("[notification-thread] Consumer sample handling error occurred");
ab0ee2ca JG	474	ret = -1;
	475	goto end;
	476	}
	477	end:
	478	return ret;
	479	}
	480
	481	/*
	482	* This thread services notification channel clients and commands received
	483	* from various lttng-sessiond components over a command queue.
	484	*/
	485	void thread_notification(void data)
	486	{
	487	int ret;
	488	struct notification_thread_handle *handle = data;
	489	struct notification_thread_state state;
	490
	491	DBG("[notification-thread] Started notification thread");
	492
	493	if (!handle) {
	494	ERR("[notification-thread] Invalid thread context provided");
	495	goto end;
	496	}
	497
	498	rcu_register_thread();
	499	rcu_thread_online();
	500
	501	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
	502	health_code_update();
	503
	504	ret = init_thread_state(handle, &state);
	505	if (ret) {
	506	goto end;
	507	}
	508
	509	/* Ready to handle client connections. */
	510	sessiond_notify_ready();
	511
	512	while (true) {
	513	int fd_count, i;
	514
	515	health_poll_entry();
	516	DBG("[notification-thread] Entering poll wait");
	517	ret = lttng_poll_wait(&state.events, -1);
	518	DBG("[notification-thread] Poll wait returned (%i)", ret);
	519	health_poll_exit();
	520	if (ret < 0) {
	521	/*
	522	* Restart interrupted system call.
	523	*/
	524	if (errno == EINTR) {
	525	continue;
	526	}
	527	ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
	528	goto error;
	529	}
	530
	531	fd_count = ret;
	532	for (i = 0; i < fd_count; i++) {
	533	int fd = LTTNG_POLL_GETFD(&state.events, i);
	534	uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
	535
2d01d977 JG	536	if (!revents) {
	537	continue;
	538	}
ab0ee2ca JG	539	DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
	540
	541	if (fd == state.notification_channel_socket) {
	542	if (revents & LPOLLIN) {
	543	ret = handle_notification_thread_client_connect(
	544	&state);
	545	if (ret < 0) {
	546	goto error;
	547	}
	548	} else if (revents &
	549	(LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	550	ERR("[notification-thread] Notification socket poll error");
	551	goto error;
	552	} else {
	553	ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
	554	goto error;
	555	}
814b4934	556	} else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
ab0ee2ca JG	557	ret = handle_notification_thread_command(handle,
	558	&state);
	559	if (ret < 0) {
	560	DBG("[notification-thread] Error encountered while servicing command queue");
	561	goto error;
	562	} else if (ret > 0) {
	563	goto exit;
	564	}
	565	} else if (fd == handle->channel_monitoring_pipes.ust32_consumer \|\|
	566	fd == handle->channel_monitoring_pipes.ust64_consumer \|\|
	567	fd == handle->channel_monitoring_pipes.kernel_consumer) {
	568	ret = handle_channel_monitoring_pipe(fd,
	569	revents, handle, &state);
	570	if (ret) {
	571	goto error;
	572	}
	573	} else {
	574	/* Activity on a client's socket. */
	575	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	576	/*
	577	* It doesn't matter if a command was
	578	* pending on the client socket at this
	579	* point since it now has no way to
	580	* receive the notifications to which
	581	* it was subscribing or unsubscribing.
	582	*/
	583	ret = handle_notification_thread_client_disconnect(
	584	fd, &state);
	585	if (ret) {
	586	goto error;
	587	}
	588	} else {
	589	if (revents & LPOLLIN) {
	590	ret = handle_notification_thread_client_in(
	591	&state, fd);
	592	if (ret) {
	593	goto error;
	594	}
	595	}
	596
	597	if (revents & LPOLLOUT) {
	598	ret = handle_notification_thread_client_out(
	599	&state, fd);
	600	if (ret) {
	601	goto error;
	602	}
	603	}
	604	}
	605	}
	606	}
	607	}
	608	exit:
	609	error:
	610	fini_thread_state(&state);
	611	health_unregister(health_sessiond);
	612	rcu_thread_offline();
	613	rcu_unregister_thread();
	614	end:
	615	return NULL;
	616	}