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