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

/*
 * Copyright (C) 2017 - Julien Desfossez <jdesfossez@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 <common/error.h>
#include <common/config/session-config.h>
#include <common/defaults.h>
#include <common/utils.h>
#include <common/futex.h>
#include <common/align.h>
#include <common/time.h>
#include <common/hashtable/utils.h>
#include <sys/eventfd.h>
#include <sys/stat.h>
#include <time.h>
#include <signal.h>
#include <inttypes.h>

#include <common/kernel-ctl/kernel-ctl.h>
#include "rotation-thread.h"
#include "lttng-sessiond.h"
#include "health-sessiond.h"
#include "rotate.h"
#include "cmd.h"
#include "sessiond-timer.h"

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

struct cds_lfht *channel_pending_rotate_ht;

static
void channel_rotation_info_destroy(struct rotation_channel_info *channel_info)
{
	assert(channel_info);
	free(channel_info);
}

static
int match_channel_info(struct cds_lfht_node *node, const void *key)
{
	struct rotation_channel_key *channel_key = (struct rotation_channel_key *) key;
	struct rotation_channel_info *channel_info;

	channel_info = caa_container_of(node, struct rotation_channel_info,
			rotate_channels_ht_node);

	return !!((channel_key->key == channel_info->channel_key.key) &&
			(channel_key->domain == channel_info->channel_key.domain));
}

static
struct rotation_channel_info *lookup_channel_pending(uint64_t key,
		enum lttng_domain_type domain)
{
	struct cds_lfht_iter iter;
	struct cds_lfht_node *node;
	struct rotation_channel_info *channel_info = NULL;
	struct rotation_channel_key channel_key = { .key = key,
		.domain = domain };

	cds_lfht_lookup(channel_pending_rotate_ht,
			hash_channel_key(&channel_key),
			match_channel_info,
			&channel_key, &iter);
	node = cds_lfht_iter_get_node(&iter);
	if (!node) {
		goto end;
	}

	channel_info = caa_container_of(node, struct rotation_channel_info,
			rotate_channels_ht_node);
	cds_lfht_del(channel_pending_rotate_ht, node);
end:
	return channel_info;
}

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

	if (!handle) {
		goto end;
	}

	if (handle->ust32_consumer >= 0) {
		ret = close(handle->ust32_consumer);
		if (ret) {
			PERROR("close 32-bit consumer channel rotation pipe");
		}
	}
	if (handle->ust64_consumer >= 0) {
		ret = close(handle->ust64_consumer);
		if (ret) {
			PERROR("close 64-bit consumer channel rotation pipe");
		}
	}
	if (handle->kernel_consumer >= 0) {
		ret = close(handle->kernel_consumer);
		if (ret) {
			PERROR("close kernel consumer channel rotation pipe");
		}
	}

end:
	free(handle);
}

struct rotation_thread_handle *rotation_thread_handle_create(
		struct lttng_pipe *ust32_channel_rotate_pipe,
		struct lttng_pipe *ust64_channel_rotate_pipe,
		struct lttng_pipe *kernel_channel_rotate_pipe,
		int thread_quit_pipe, int rotate_timer_pipe)
{
	struct rotation_thread_handle *handle;

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

	if (ust32_channel_rotate_pipe) {
		handle->ust32_consumer =
				lttng_pipe_release_readfd(
					ust32_channel_rotate_pipe);
		if (handle->ust32_consumer < 0) {
			goto error;
		}
	} else {
		handle->ust32_consumer = -1;
	}
	if (ust64_channel_rotate_pipe) {
		handle->ust64_consumer =
				lttng_pipe_release_readfd(
					ust64_channel_rotate_pipe);
		if (handle->ust64_consumer < 0) {
			goto error;
		}
	} else {
		handle->ust64_consumer = -1;
	}
	if (kernel_channel_rotate_pipe) {
		handle->kernel_consumer =
				lttng_pipe_release_readfd(
					kernel_channel_rotate_pipe);
		if (handle->kernel_consumer < 0) {
			goto error;
		}
	} else {
		handle->kernel_consumer = -1;
	}
	handle->thread_quit_pipe = thread_quit_pipe;
	handle->rotate_timer_pipe = rotate_timer_pipe;

end:
	return handle;
error:
	rotation_thread_handle_destroy(handle);
	return NULL;
}

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

	/*
	 * Create pollset with size 5:
	 *	- sessiond quit pipe
	 *	- consumerd (32-bit user space) channel rotate pipe,
	 *	- consumerd (64-bit user space) channel rotate pipe,
	 *	- consumerd (kernel) channel rotate pipe.
	 *	- sessiond rotate pending pipe
	 */
	ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
	if (ret < 0) {
		goto end;
	}

	ret = lttng_poll_add(poll_set, handle->thread_quit_pipe,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[rotation-thread] Failed to add thread_quit_pipe fd to pollset");
		goto error;
	}
	ret = lttng_poll_add(poll_set, handle->rotate_timer_pipe,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[rotation-thread] Failed to add rotate_pending fd to pollset");
		goto error;
	}
	ret = lttng_poll_add(poll_set, handle->ust32_consumer,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[rotation-thread] Failed to add ust-32 channel rotation pipe fd to pollset");
		goto error;
	}
	ret = lttng_poll_add(poll_set, handle->ust64_consumer,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[rotation-thread] Failed to add ust-64 channel rotation pipe fd to pollset");
		goto error;
	}
	if (handle->kernel_consumer < 0) {
		goto end;
	}
	ret = lttng_poll_add(poll_set, handle->kernel_consumer,
			LPOLLIN | LPOLLERR);
	if (ret < 0) {
		ERR("[rotation-thread] Failed to add kernel channel rotation pipe fd to pollset");
		goto error;
	}

end:
	return ret;
error:
	lttng_poll_clean(poll_set);
	return ret;
}

static
void fini_thread_state(struct rotation_thread_state *state)
{
	lttng_poll_clean(&state->events);
}

static
int init_thread_state(struct rotation_thread_handle *handle,
		struct rotation_thread_state *state)
{
	int ret;

	memset(state, 0, sizeof(*state));
	lttng_poll_init(&state->events);

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

	channel_pending_rotate_ht = cds_lfht_new(DEFAULT_HT_SIZE,
			1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
	if (!channel_pending_rotate_ht) {
		ret = -1;
	}

end:
	return 0;
}

static
int handle_channel_rotation_pipe(int fd, uint32_t revents,
		struct rotation_thread_handle *handle,
		struct rotation_thread_state *state)
{
	int ret = 0;
	enum lttng_domain_type domain;
	struct rotation_channel_info *channel_info;
	uint64_t key;

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

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

	do {
		ret = read(fd, &key, sizeof(key));
	} while (ret == -1 && errno == EINTR);
	if (ret != sizeof(key)) {
		ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
				fd);
		ret = -1;
		goto end;
	}

	DBG("[rotation-thread] Received notification for chan %" PRIu64
			", domain %d\n", key, domain);

	channel_info = lookup_channel_pending(key, domain);
	if (!channel_info) {
		ERR("[rotation-thread] Failed to find channel_info (key = %"
				PRIu64 ")", key);
		ret = -1;
		goto end;
	}

	if (--channel_info->session->nr_chan_rotate_pending == 0) {
		time_t now = time(NULL);

		if (now == (time_t) -1) {
			channel_info->session->rotate_status = LTTNG_ROTATE_ERROR;
			ret = LTTNG_ERR_ROTATE_NOT_AVAILABLE;
			goto end;
		}

		ret = rename_complete_chunk(channel_info->session, now);
		if (ret < 0) {
			ERR("Failed to rename completed rotation chunk");
			goto end;
		}
		channel_info->session->rotate_pending = false;
		if (channel_info->session->rotate_pending_relay) {
			ret = sessiond_timer_rotate_pending_start(
					channel_info->session,
					DEFAULT_ROTATE_PENDING_RELAY_TIMER);
			if (ret) {
				ERR("Enabling rotate pending timer");
				ret = -1;
				goto end;
			}
		}
	}

	channel_rotation_info_destroy(channel_info);

	ret = 0;

end:
	return ret;
}

static
int rotate_pending_relay_timer(struct ltt_session *session)
{
	int ret;

	DBG("[rotation-thread] Check rotate pending on session %" PRIu64,
			session->id);
	ret = relay_rotate_pending(session, session->rotate_count - 1);
	if (ret < 0) {
		ERR("[rotation-thread] Check relay rotate pending");
		goto end;
	}
	if (ret == 0) {
		DBG("[rotation-thread] Rotation completed on the relay for "
				"session %" PRIu64, session->id);
		/*
		 * Stop the timer and clear the queue, the timers are currently
		 * ignored because of the rotate_pending_relay_check_in_progress
		 * flag.
		 */
		sessiond_timer_rotate_pending_stop(session);
		/*
		 * Now we can clear the pending flag in the session. New
		 * rotations can start now.
		 */
		session->rotate_pending_relay = false;
	} else if (ret == 1) {
		DBG("[rotation-thread] Rotation still pending on the relay for "
				"session %" PRIu64, session->id);
	}
	/*
	 * Allow the timer thread to send other notifications when needed.
	 */
	session->rotate_pending_relay_check_in_progress = false;
	fprintf(stderr, "RET PENDING: %d\n", ret);

	ret = 0;

end:
	return ret;
}

static
int rotate_timer(struct ltt_session *session)
{
	int ret;

	DBG("[rotation-thread] Rotate timer on session %" PRIu64, session->id);

	/*
	 * If the session is stopped, we need to cancel this timer.
	 */
	session_lock(session);
	if (!session->active && session->rotate_timer_enabled) {
		sessiond_rotate_timer_stop(session);
	}

	ret = cmd_rotate_session(session, NULL);
	session_unlock(session);
	fprintf(stderr, "RET ROTATE TIMER: %d\n", ret);
	if (ret == -LTTNG_ERR_ROTATE_PENDING) {
		ret = 0;
		goto end;
	} else if (ret != LTTNG_OK) {
		ret = -1;
		goto end;
	}


	ret = 0;

end:
	return ret;
}

static
int handle_rotate_timer_pipe(int fd, uint32_t revents,
		struct rotation_thread_handle *handle,
		struct rotation_thread_state *state)
{
	int ret = 0;
	struct ltt_session *session;
	struct sessiond_rotation_timer timer_data;

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

	memset(&timer_data, 0, sizeof(struct sessiond_rotation_timer));

	do {
		ret = read(fd, &timer_data, sizeof(timer_data));
	} while (ret == -1 && errno == EINTR);
	if (ret != sizeof(timer_data)) {
		ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
				fd);
		ret = -1;
		goto end;
	}

	rcu_read_lock();
	session_lock_list();
	session = session_find_by_id(timer_data.session_id);
	if (!session) {
		ERR("[rotation-thread] Session %" PRIu64 " not found",
				timer_data.session_id);
		ret = -1;
		goto end_unlock;
	}

	if (timer_data.signal == LTTNG_SESSIOND_SIG_ROTATE_PENDING) {
		ret = rotate_pending_relay_timer(session);
	} else if (timer_data.signal == LTTNG_SESSIOND_SIG_ROTATE_TIMER) {
		ret = rotate_timer(session);
	} else {
		ERR("Unknown signal in rotate timer");
		ret = -1;
	}

end_unlock:
	session_unlock_list();
	rcu_read_unlock();
end:
	return ret;
}

void *thread_rotation(void *data)
{
	int ret;
	struct rotation_thread_handle *handle = data;
	struct rotation_thread_state state;

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

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

	rcu_register_thread();
	rcu_thread_online();

	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
	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("[rotation-thread] Entering poll wait");
		ret = lttng_poll_wait(&state.events, -1);
		DBG("[rotation-thread] Poll wait returned (%i)", ret);
		health_poll_exit();
		if (ret < 0) {
			/*
			 * Restart interrupted system call.
			 */
			if (errno == EINTR) {
				continue;
			}
			ERR("[rotation-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("[rotation-thread] Handling fd (%i) activity (%u)",
					fd, revents);

			if (fd == handle->thread_quit_pipe) {
				DBG("[rotation-thread] Quit pipe activity");
				goto exit;
			} else if (fd == handle->rotate_timer_pipe) {
				ret = handle_rotate_timer_pipe(fd, revents,
						handle, &state);
				if (ret) {
					ERR("[rotation-thread] Rotate timer");
					goto error;
				}
			} else if (fd == handle->ust32_consumer ||
					fd == handle->ust64_consumer ||
					fd == handle->kernel_consumer) {
				ret = handle_channel_rotation_pipe(fd,
						revents, handle, &state);
				if (ret) {
					ERR("[rotation-thread] Exit main loop");
					goto error;
				}
			}
		}
	}
exit:
error:
	fini_thread_state(&state);
	health_unregister(health_sessiond);
	rcu_thread_offline();
	rcu_unregister_thread();
end:
	return NULL;
}
Commit	Line	Data
6e0be6cc JD	1	/*
	2	* Copyright (C) 2017 - Julien Desfossez <jdesfossez@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 <common/error.h>
	21	#include <common/config/session-config.h>
	22	#include <common/defaults.h>
	23	#include <common/utils.h>
	24	#include <common/futex.h>
	25	#include <common/align.h>
	26	#include <common/time.h>
	27	#include <common/hashtable/utils.h>
	28	#include <sys/eventfd.h>
	29	#include <sys/stat.h>
	30	#include <time.h>
	31	#include <signal.h>
	32	#include <inttypes.h>
	33
	34	#include <common/kernel-ctl/kernel-ctl.h>
	35	#include "rotation-thread.h"
	36	#include "lttng-sessiond.h"
	37	#include "health-sessiond.h"
e8ed1ed1	38	#include "rotate.h"
6e0be6cc	39	#include "cmd.h"
0d2ff20e	40	#include "sessiond-timer.h"
6e0be6cc JD	41
	42	#include <urcu.h>
	43	#include <urcu/list.h>
	44	#include <urcu/rculfhash.h>
	45
	46	struct cds_lfht *channel_pending_rotate_ht;
	47
6e0be6cc JD	48	static
	49	void channel_rotation_info_destroy(struct rotation_channel_info *channel_info)
	50	{
	51	assert(channel_info);
	52	free(channel_info);
	53	}
	54
6e0be6cc JD	55	static
	56	int match_channel_info(struct cds_lfht_node node, const void key)
	57	{
	58	struct rotation_channel_key channel_key = (struct rotation_channel_key ) key;
	59	struct rotation_channel_info *channel_info;
	60
	61	channel_info = caa_container_of(node, struct rotation_channel_info,
	62	rotate_channels_ht_node);
	63
	64	return !!((channel_key->key == channel_info->channel_key.key) &&
	65	(channel_key->domain == channel_info->channel_key.domain));
	66	}
	67
	68	static
	69	struct rotation_channel_info *lookup_channel_pending(uint64_t key,
	70	enum lttng_domain_type domain)
	71	{
	72	struct cds_lfht_iter iter;
	73	struct cds_lfht_node *node;
	74	struct rotation_channel_info *channel_info = NULL;
	75	struct rotation_channel_key channel_key = { .key = key,
	76	.domain = domain };
	77
	78	cds_lfht_lookup(channel_pending_rotate_ht,
	79	hash_channel_key(&channel_key),
	80	match_channel_info,
	81	&channel_key, &iter);
	82	node = cds_lfht_iter_get_node(&iter);
	83	if (!node) {
	84	goto end;
	85	}
	86
	87	channel_info = caa_container_of(node, struct rotation_channel_info,
	88	rotate_channels_ht_node);
	89	cds_lfht_del(channel_pending_rotate_ht, node);
	90	end:
	91	return channel_info;
	92	}
	93
	94	/*
	95	* Destroy the thread data previously created by the init function.
	96	*/
	97	void rotation_thread_handle_destroy(
	98	struct rotation_thread_handle *handle)
	99	{
	100	int ret;
	101
	102	if (!handle) {
	103	goto end;
	104	}
	105
	106	if (handle->ust32_consumer >= 0) {
	107	ret = close(handle->ust32_consumer);
	108	if (ret) {
	109	PERROR("close 32-bit consumer channel rotation pipe");
	110	}
	111	}
	112	if (handle->ust64_consumer >= 0) {
	113	ret = close(handle->ust64_consumer);
	114	if (ret) {
	115	PERROR("close 64-bit consumer channel rotation pipe");
	116	}
	117	}
	118	if (handle->kernel_consumer >= 0) {
119	ret = close(handle->kernel_consumer);
120	if (ret) {
121	PERROR("close kernel consumer channel rotation pipe");
122	}
123	}
124
125	end:
126	free(handle);
127	}
128
129	struct rotation_thread_handle *rotation_thread_handle_create(
130	struct lttng_pipe *ust32_channel_rotate_pipe,
131	struct lttng_pipe *ust64_channel_rotate_pipe,
132	struct lttng_pipe *kernel_channel_rotate_pipe,
55c2a7f9	133	int thread_quit_pipe, int rotate_timer_pipe)
6e0be6cc JD	134	{
	135	struct rotation_thread_handle *handle;
	136
	137	handle = zmalloc(sizeof(*handle));
	138	if (!handle) {
	139	goto end;
	140	}
	141
	142	if (ust32_channel_rotate_pipe) {
	143	handle->ust32_consumer =
	144	lttng_pipe_release_readfd(
	145	ust32_channel_rotate_pipe);
	146	if (handle->ust32_consumer < 0) {
	147	goto error;
	148	}
	149	} else {
	150	handle->ust32_consumer = -1;
	151	}
	152	if (ust64_channel_rotate_pipe) {
	153	handle->ust64_consumer =
	154	lttng_pipe_release_readfd(
	155	ust64_channel_rotate_pipe);
	156	if (handle->ust64_consumer < 0) {
	157	goto error;
	158	}
	159	} else {
	160	handle->ust64_consumer = -1;
	161	}
	162	if (kernel_channel_rotate_pipe) {
	163	handle->kernel_consumer =
	164	lttng_pipe_release_readfd(
	165	kernel_channel_rotate_pipe);
	166	if (handle->kernel_consumer < 0) {
	167	goto error;
	168	}
	169	} else {
	170	handle->kernel_consumer = -1;
	171	}
	172	handle->thread_quit_pipe = thread_quit_pipe;
55c2a7f9	173	handle->rotate_timer_pipe = rotate_timer_pipe;
6e0be6cc JD	174
	175	end:
	176	return handle;
	177	error:
	178	rotation_thread_handle_destroy(handle);
	179	return NULL;
	180	}
	181
	182	static
	183	int init_poll_set(struct lttng_poll_event *poll_set,
	184	struct rotation_thread_handle *handle)
	185	{
	186	int ret;
	187
	188	/*
55c2a7f9	189	* Create pollset with size 5:
6e0be6cc JD	190	* - sessiond quit pipe
	191	* - consumerd (32-bit user space) channel rotate pipe,
	192	* - consumerd (64-bit user space) channel rotate pipe,
	193	* - consumerd (kernel) channel rotate pipe.
55c2a7f9	194	* - sessiond rotate pending pipe
6e0be6cc	195	*/
55c2a7f9	196	ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
6e0be6cc JD	197	if (ret < 0) {
	198	goto end;
	199	}
	200
	201	ret = lttng_poll_add(poll_set, handle->thread_quit_pipe,
	202	LPOLLIN \| LPOLLERR);
	203	if (ret < 0) {
	204	ERR("[rotation-thread] Failed to add thread_quit_pipe fd to pollset");
	205	goto error;
	206	}
55c2a7f9 JD	207	ret = lttng_poll_add(poll_set, handle->rotate_timer_pipe,
	208	LPOLLIN \| LPOLLERR);
	209	if (ret < 0) {
	210	ERR("[rotation-thread] Failed to add rotate_pending fd to pollset");
	211	goto error;
	212	}
6e0be6cc JD	213	ret = lttng_poll_add(poll_set, handle->ust32_consumer,
	214	LPOLLIN \| LPOLLERR);
	215	if (ret < 0) {
	216	ERR("[rotation-thread] Failed to add ust-32 channel rotation pipe fd to pollset");
	217	goto error;
	218	}
	219	ret = lttng_poll_add(poll_set, handle->ust64_consumer,
	220	LPOLLIN \| LPOLLERR);
	221	if (ret < 0) {
	222	ERR("[rotation-thread] Failed to add ust-64 channel rotation pipe fd to pollset");
	223	goto error;
	224	}
	225	if (handle->kernel_consumer < 0) {
	226	goto end;
	227	}
	228	ret = lttng_poll_add(poll_set, handle->kernel_consumer,
	229	LPOLLIN \| LPOLLERR);
	230	if (ret < 0) {
	231	ERR("[rotation-thread] Failed to add kernel channel rotation pipe fd to pollset");
	232	goto error;
	233	}
	234
	235	end:
	236	return ret;
	237	error:
	238	lttng_poll_clean(poll_set);
	239	return ret;
	240	}
	241
	242	static
	243	void fini_thread_state(struct rotation_thread_state *state)
	244	{
	245	lttng_poll_clean(&state->events);
	246	}
	247
	248	static
	249	int init_thread_state(struct rotation_thread_handle *handle,
	250	struct rotation_thread_state *state)
	251	{
	252	int ret;
	253
	254	memset(state, 0, sizeof(*state));
	255	lttng_poll_init(&state->events);
	256
	257	ret = init_poll_set(&state->events, handle);
	258	if (ret) {
	259	goto end;
	260	}
	261
	262	channel_pending_rotate_ht = cds_lfht_new(DEFAULT_HT_SIZE,
	263	1, 0, CDS_LFHT_AUTO_RESIZE \| CDS_LFHT_ACCOUNTING, NULL);
	264	if (!channel_pending_rotate_ht) {
	265	ret = -1;
	266	}
	267
	268	end:
	269	return 0;
	270	}
	271
6e0be6cc JD	272	static
	273	int handle_channel_rotation_pipe(int fd, uint32_t revents,
	274	struct rotation_thread_handle *handle,
	275	struct rotation_thread_state *state)
	276	{
	277	int ret = 0;
	278	enum lttng_domain_type domain;
	279	struct rotation_channel_info *channel_info;
	280	uint64_t key;
	281
	282	if (fd == handle->ust32_consumer \|\|
	283	fd == handle->ust64_consumer) {
	284	domain = LTTNG_DOMAIN_UST;
	285	} else if (fd == handle->kernel_consumer) {
	286	domain = LTTNG_DOMAIN_KERNEL;
	287	} else {
	288	abort();
	289	}
	290
	291	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	292	ret = lttng_poll_del(&state->events, fd);
	293	if (ret) {
103373d7 JD	294	ERR("[rotation-thread] Failed to remove consumer "
103373d7 JD	295	"rotation pipe from poll set");
6e0be6cc JD	296	}
	297	goto end;
	298	}
	299
	300	do {
	301	ret = read(fd, &key, sizeof(key));
	302	} while (ret == -1 && errno == EINTR);
	303	if (ret != sizeof(key)) {
	304	ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
	305	fd);
	306	ret = -1;
	307	goto end;
	308	}
	309
	310	DBG("[rotation-thread] Received notification for chan %" PRIu64
	311	", domain %d\n", key, domain);
	312
	313	channel_info = lookup_channel_pending(key, domain);
	314	if (!channel_info) {
	315	ERR("[rotation-thread] Failed to find channel_info (key = %"
	316	PRIu64 ")", key);
	317	ret = -1;
	318	goto end;
	319	}
	320
	321	if (--channel_info->session->nr_chan_rotate_pending == 0) {
	322	time_t now = time(NULL);
	323
	324	if (now == (time_t) -1) {
4a478f45	325	channel_info->session->rotate_status = LTTNG_ROTATE_ERROR;
6e0be6cc JD	326	ret = LTTNG_ERR_ROTATE_NOT_AVAILABLE;
	327	goto end;
	328	}
	329
	330	ret = rename_complete_chunk(channel_info->session, now);
	331	if (ret < 0) {
	332	ERR("Failed to rename completed rotation chunk");
	333	goto end;
	334	}
d3dedf27	335	channel_info->session->rotate_pending = false;
0d2ff20e JD	336	if (channel_info->session->rotate_pending_relay) {
0d2ff20e JD	337	ret = sessiond_timer_rotate_pending_start(
1b58b7f5 JD	338	channel_info->session,
1b58b7f5 JD	339	DEFAULT_ROTATE_PENDING_RELAY_TIMER);
0d2ff20e JD	340	if (ret) {
	341	ERR("Enabling rotate pending timer");
	342	ret = -1;
	343	goto end;
	344	}
	345	}
6e0be6cc JD	346	}
	347
	348	channel_rotation_info_destroy(channel_info);
	349
	350	ret = 0;
	351
	352	end:
	353	return ret;
	354	}
	355
06e13874 JD	356	static
	357	int rotate_pending_relay_timer(struct ltt_session *session)
	358	{
	359	int ret;
	360
	361	DBG("[rotation-thread] Check rotate pending on session %" PRIu64,
	362	session->id);
	363	ret = relay_rotate_pending(session, session->rotate_count - 1);
	364	if (ret < 0) {
	365	ERR("[rotation-thread] Check relay rotate pending");
	366	goto end;
	367	}
	368	if (ret == 0) {
	369	DBG("[rotation-thread] Rotation completed on the relay for "
	370	"session %" PRIu64, session->id);
	371	/*
	372	* Stop the timer and clear the queue, the timers are currently
	373	* ignored because of the rotate_pending_relay_check_in_progress
	374	* flag.
	375	*/
	376	sessiond_timer_rotate_pending_stop(session);
	377	/*
	378	* Now we can clear the pending flag in the session. New
	379	* rotations can start now.
	380	*/
	381	session->rotate_pending_relay = false;
	382	} else if (ret == 1) {
	383	DBG("[rotation-thread] Rotation still pending on the relay for "
	384	"session %" PRIu64, session->id);
	385	}
	386	/*
	387	* Allow the timer thread to send other notifications when needed.
	388	*/
	389	session->rotate_pending_relay_check_in_progress = false;
	390	fprintf(stderr, "RET PENDING: %d\n", ret);
	391
	392	ret = 0;
	393
	394	end:
	395	return ret;
	396	}
	397
	398	static
	399	int rotate_timer(struct ltt_session *session)
	400	{
	401	int ret;
	402
	403	DBG("[rotation-thread] Rotate timer on session %" PRIu64, session->id);
8cf2762e JD	404
	405	/*
	406	* If the session is stopped, we need to cancel this timer.
	407	*/
90097149	408	session_lock(session);
8cf2762e JD	409	if (!session->active && session->rotate_timer_enabled) {
	410	sessiond_rotate_timer_stop(session);
	411	}
8cf2762e	412
06e13874	413	ret = cmd_rotate_session(session, NULL);
90097149	414	session_unlock(session);
06e13874	415	fprintf(stderr, "RET ROTATE TIMER: %d\n", ret);
8cf2762e JD	416	if (ret == -LTTNG_ERR_ROTATE_PENDING) {
	417	ret = 0;
	418	goto end;
	419	} else if (ret != LTTNG_OK) {
	420	ret = -1;
	421	goto end;
	422	}
	423
06e13874	424
8cf2762e JD	425	ret = 0;
	426
	427	end:
06e13874 JD	428	return ret;
	429	}
	430
55c2a7f9 JD	431	static
	432	int handle_rotate_timer_pipe(int fd, uint32_t revents,
	433	struct rotation_thread_handle *handle,
	434	struct rotation_thread_state *state)
	435	{
	436	int ret = 0;
55c2a7f9	437	struct ltt_session *session;
06e13874	438	struct sessiond_rotation_timer timer_data;
55c2a7f9 JD	439
	440	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	441	ret = lttng_poll_del(&state->events, fd);
	442	if (ret) {
	443	ERR("[rotation-thread] Failed to remove consumer "
	444	"rotate pending pipe from poll set");
	445	}
	446	goto end;
	447	}
	448
06e13874 JD	449	memset(&timer_data, 0, sizeof(struct sessiond_rotation_timer));
06e13874 JD	450
55c2a7f9	451	do {
06e13874	452	ret = read(fd, &timer_data, sizeof(timer_data));
55c2a7f9	453	} while (ret == -1 && errno == EINTR);
06e13874	454	if (ret != sizeof(timer_data)) {
55c2a7f9 JD	455	ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
	456	fd);
	457	ret = -1;
	458	goto end;
	459	}
	460
90097149 JD	461	rcu_read_lock();
90097149 JD	462	session_lock_list();
06e13874	463	session = session_find_by_id(timer_data.session_id);
55c2a7f9 JD	464	if (!session) {
55c2a7f9 JD	465	ERR("[rotation-thread] Session %" PRIu64 " not found",
06e13874	466	timer_data.session_id);
55c2a7f9	467	ret = -1;
90097149	468	goto end_unlock;
55c2a7f9 JD	469	}
55c2a7f9 JD	470
06e13874 JD	471	if (timer_data.signal == LTTNG_SESSIOND_SIG_ROTATE_PENDING) {
	472	ret = rotate_pending_relay_timer(session);
	473	} else if (timer_data.signal == LTTNG_SESSIOND_SIG_ROTATE_TIMER) {
	474	ret = rotate_timer(session);
	475	} else {
	476	ERR("Unknown signal in rotate timer");
	477	ret = -1;
55c2a7f9	478	}
55c2a7f9	479
90097149 JD	480	end_unlock:
	481	session_unlock_list();
	482	rcu_read_unlock();
55c2a7f9 JD	483	end:
	484	return ret;
	485	}
	486
6e0be6cc JD	487	void thread_rotation(void data)
	488	{
	489	int ret;
	490	struct rotation_thread_handle *handle = data;
	491	struct rotation_thread_state state;
	492
	493	DBG("[rotation-thread] Started rotation thread");
	494
	495	if (!handle) {
	496	ERR("[rotation-thread] Invalid thread context provided");
	497	goto end;
	498	}
	499
	500	rcu_register_thread();
	501	rcu_thread_online();
	502
	503	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
	504	health_code_update();
	505
	506	ret = init_thread_state(handle, &state);
	507	if (ret) {
	508	goto end;
	509	}
	510
	511	/* Ready to handle client connections. */
	512	sessiond_notify_ready();
	513
	514	while (true) {
	515	int fd_count, i;
	516
	517	health_poll_entry();
	518	DBG("[rotation-thread] Entering poll wait");
	519	ret = lttng_poll_wait(&state.events, -1);
	520	DBG("[rotation-thread] Poll wait returned (%i)", ret);
	521	health_poll_exit();
	522	if (ret < 0) {
	523	/*
	524	* Restart interrupted system call.
	525	*/
	526	if (errno == EINTR) {
	527	continue;
	528	}
	529	ERR("[rotation-thread] Error encountered during lttng_poll_wait (%i)", ret);
	530	goto error;
	531	}
	532
	533	fd_count = ret;
	534	for (i = 0; i < fd_count; i++) {
	535	int fd = LTTNG_POLL_GETFD(&state.events, i);
	536	uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
	537
	538	DBG("[rotation-thread] Handling fd (%i) activity (%u)",
	539	fd, revents);
	540
	541	if (fd == handle->thread_quit_pipe) {
	542	DBG("[rotation-thread] Quit pipe activity");
	543	goto exit;
55c2a7f9 JD	544	} else if (fd == handle->rotate_timer_pipe) {
	545	ret = handle_rotate_timer_pipe(fd, revents,
	546	handle, &state);
	547	if (ret) {
06e13874	548	ERR("[rotation-thread] Rotate timer");
55c2a7f9 JD	549	goto error;
55c2a7f9 JD	550	}
6e0be6cc JD	551	} else if (fd == handle->ust32_consumer \|\|
	552	fd == handle->ust64_consumer \|\|
	553	fd == handle->kernel_consumer) {
	554	ret = handle_channel_rotation_pipe(fd,
	555	revents, handle, &state);
	556	if (ret) {
	557	ERR("[rotation-thread] Exit main loop");
	558	goto error;
	559	}
	560	}
	561	}
	562	}
	563	exit:
	564	error:
	565	fini_thread_state(&state);
	566	health_unregister(health_sessiond);
	567	rcu_thread_offline();
	568	rcu_unregister_thread();
	569	end:
	570	return NULL;
	571	}