[deliverable/lttng-tools.git] / src / bin / lttng-sessiond / sessiond-timer.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 <assert.h>
#include <inttypes.h>
#include <signal.h>

#include "sessiond-timer.h"
#include "health-sessiond.h"

static struct timer_signal_data timer_signal = {
	.tid = 0,
	.setup_done = 0,
	.qs_done = 0,
	.lock = PTHREAD_MUTEX_INITIALIZER,
};

/*
 * Set custom signal mask to current thread.
 */
static void setmask(sigset_t *mask)
{
	int ret;

	ret = sigemptyset(mask);
	if (ret) {
		PERROR("sigemptyset");
	}
	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_TEARDOWN);
	if (ret) {
		PERROR("sigaddset teardown");
	}
	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_EXIT);
	if (ret) {
		PERROR("sigaddset exit");
	}
	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_ROTATE_PENDING);
	if (ret) {
		PERROR("sigaddset switch");
	}
	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_ROTATE_TIMER);
	if (ret) {
		PERROR("sigaddset switch");
	}
}

static
void sessiond_timer_signal_thread_qs(unsigned int signr)
{
	sigset_t pending_set;
	int ret;

	/*
	 * We need to be the only thread interacting with the thread
	 * that manages signals for teardown synchronization.
	 */
	pthread_mutex_lock(&timer_signal.lock);

	/* Ensure we don't have any signal queued for this session. */
	for (;;) {
		ret = sigemptyset(&pending_set);
		if (ret == -1) {
			PERROR("sigemptyset");
		}
		ret = sigpending(&pending_set);
		if (ret == -1) {
			PERROR("sigpending");
		}
		if (!sigismember(&pending_set, signr)) {
			break;
		}
		caa_cpu_relax();
	}

	/*
	 * From this point, no new signal handler will be fired that would try to
	 * access "session". However, we still need to wait for any currently
	 * executing handler to complete.
	 */
	cmm_smp_mb();
	CMM_STORE_SHARED(timer_signal.qs_done, 0);
	cmm_smp_mb();

	/*
	 * Kill with LTTNG_SESSIOND_SIG_TEARDOWN, so signal management thread wakes
	 * up.
	 */
	kill(getpid(), LTTNG_SESSIOND_SIG_TEARDOWN);

	while (!CMM_LOAD_SHARED(timer_signal.qs_done)) {
		caa_cpu_relax();
	}
	cmm_smp_mb();

	pthread_mutex_unlock(&timer_signal.lock);
}

/*
 * Start a timer on a session that will fire at a given interval
 * (timer_interval_us) and fire a given signal (signal).
 *
 * Returns a negative value on error, 0 if a timer was created, and
 * a positive value if no timer was created (not an error).
 */
static
int session_timer_start(timer_t *timer_id, struct ltt_session *session,
		unsigned int timer_interval_us, int signal)
{
	int ret = 0, delete_ret;
	struct sigevent sev;
	struct itimerspec its;

	assert(session);

	sev.sigev_notify = SIGEV_SIGNAL;
	sev.sigev_signo = signal;
	sev.sigev_value.sival_ptr = session;
	ret = timer_create(CLOCKID, &sev, timer_id);
	if (ret == -1) {
		PERROR("timer_create");
		goto end;
	}

	its.it_value.tv_sec = timer_interval_us / 1000000;
	its.it_value.tv_nsec = (timer_interval_us % 1000000) * 1000;
	its.it_interval.tv_sec = its.it_value.tv_sec;
	its.it_interval.tv_nsec = its.it_value.tv_nsec;

	ret = timer_settime(*timer_id, 0, &its, NULL);
	if (ret == -1) {
		PERROR("timer_settime");
		goto error_destroy_timer;
	}
	goto end;

error_destroy_timer:
	delete_ret = timer_delete(*timer_id);
	if (delete_ret == -1) {
		PERROR("timer_delete");
	}

end:
	return ret;
}


static
int session_timer_stop(timer_t *timer_id, int signal)
{
	int ret = 0;

	ret = timer_delete(*timer_id);
	if (ret == -1) {
		PERROR("timer_delete");
		goto end;
	}

	sessiond_timer_signal_thread_qs(signal);
	*timer_id = 0;
end:
	return ret;
}

int sessiond_timer_rotate_pending_start(struct ltt_session *session,
		unsigned int interval_us)
{
	int ret;

	ret = session_timer_start(&session->rotate_relay_pending_timer,
			session, interval_us,
			LTTNG_SESSIOND_SIG_ROTATE_PENDING);
	if (ret == 0) {
		session->rotate_relay_pending_timer_enabled = true;
	}

	return ret;
}

/*
 * Stop and delete the channel's live timer.
 */
void sessiond_timer_rotate_pending_stop(struct ltt_session *session)
{
	int ret;

	assert(session);

	ret = session_timer_stop(&session->rotate_relay_pending_timer,
			LTTNG_SESSIOND_SIG_ROTATE_PENDING);
	if (ret == -1) {
		ERR("Failed to stop live timer");
	}

	session->rotate_relay_pending_timer_enabled = false;
}

int sessiond_rotate_timer_start(struct ltt_session *session,
		unsigned int interval_us)
{
	int ret;

	ret = session_timer_start(&session->rotate_timer, session, interval_us,
			LTTNG_SESSIOND_SIG_ROTATE_TIMER);
	if (ret == 0) {
		session->rotate_timer_enabled = true;
	}

	return ret;
}

/*
 * Stop and delete the channel's live timer.
 */
void sessiond_rotate_timer_stop(struct ltt_session *session)
{
	int ret;

	assert(session);

	ret = session_timer_stop(&session->rotate_timer,
			LTTNG_SESSIOND_SIG_ROTATE_TIMER);
	if (ret == -1) {
		ERR("Failed to stop live timer");
	}

	session->rotate_timer_enabled = false;
}

/*
 * Block the RT signals for the entire process. It must be called from the
 * sessiond main before creating the threads
 */
int sessiond_timer_signal_init(void)
{
	int ret;
	sigset_t mask;

	/* Block signal for entire process, so only our thread processes it. */
	setmask(&mask);
	ret = pthread_sigmask(SIG_BLOCK, &mask, NULL);
	if (ret) {
		errno = ret;
		PERROR("pthread_sigmask");
		return -1;
	}
	return 0;
}

static
void relay_rotation_pending_timer(struct timer_thread_parameters *ctx,
		int sig, siginfo_t *si)
{
	int ret;
	struct ltt_session *session = si->si_value.sival_ptr;
	struct sessiond_rotation_timer timer_data;
	assert(session);

	/*
	 * Avoid sending too many requests in case the relay is slower to
	 * respond than the timer period.
	 */
	if (session->rotate_pending_relay_check_in_progress ||
			!session->rotate_pending_relay) {
		goto end;
	}

	session->rotate_pending_relay_check_in_progress = true;
	memset(&timer_data, 0, sizeof(struct sessiond_rotation_timer));
	timer_data.session_id = session->id;
	timer_data.signal = LTTNG_SESSIOND_SIG_ROTATE_PENDING;
	ret = lttng_write(ctx->rotate_timer_pipe, &timer_data,
			sizeof(timer_data));
	if (ret < sizeof(session->id)) {
		PERROR("wakeup rotate pipe");
	}

end:
	return;
}

static
void rotate_timer(struct timer_thread_parameters *ctx, int sig, siginfo_t *si)
{
	int ret;
	struct ltt_session *session = si->si_value.sival_ptr;
	struct sessiond_rotation_timer timer_data;
	assert(session);

	/*
	 * No rate limiting here, so if the timer fires too quickly, there will
	 * be a backlog of timers queued up and we will try to catch up.
	 */
	memset(&timer_data, 0, sizeof(struct sessiond_rotation_timer));
	timer_data.session_id = session->id;
	timer_data.signal = LTTNG_SESSIOND_SIG_ROTATE_TIMER;
	ret = lttng_write(ctx->rotate_timer_pipe, &timer_data,
			sizeof(timer_data));
	if (ret < sizeof(session->id)) {
		PERROR("wakeup rotate pipe");
	}

	return;
}

/*
 * This thread is the sighandler for the timer signals.
 */
void *sessiond_timer_thread(void *data)
{
	int signr;
	sigset_t mask;
	siginfo_t info;
	struct timer_thread_parameters *ctx = data;

	rcu_register_thread();
	rcu_thread_online();

	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);

	health_code_update();

	/* Only self thread will receive signal mask. */
	setmask(&mask);
	CMM_STORE_SHARED(timer_signal.tid, pthread_self());

	while (1) {
		health_code_update();

		health_poll_entry();
		signr = sigwaitinfo(&mask, &info);
		health_poll_exit();

		/*
		 * NOTE: cascading conditions are used instead of a switch case
		 * since the use of SIGRTMIN in the definition of the signals'
		 * values prevents the reduction to an integer constant.
		 */
		if (signr == -1) {
			if (errno != EINTR) {
				PERROR("sigwaitinfo");
			}
			continue;
		} else if (signr == LTTNG_SESSIOND_SIG_TEARDOWN) {
			fprintf(stderr, "TEARDOWN\n");
			cmm_smp_mb();
			CMM_STORE_SHARED(timer_signal.qs_done, 1);
			cmm_smp_mb();
			DBG("Signal timer metadata thread teardown");
		} else if (signr == LTTNG_SESSIOND_SIG_EXIT) {
			fprintf(stderr, "KILL\n");
			goto end;
		} else if (signr == LTTNG_SESSIOND_SIG_ROTATE_PENDING) {
			fprintf(stderr, "PENDING TIMER\n");
			relay_rotation_pending_timer(ctx, info.si_signo, &info);
		} else if (signr == LTTNG_SESSIOND_SIG_ROTATE_TIMER) {
			fprintf(stderr, "ROTATE TIMER\n");
			rotate_timer(ctx, info.si_signo, &info);
		} else {
			ERR("Unexpected signal %d\n", info.si_signo);
		}
	}

end:
	health_unregister(health_sessiond);
	rcu_thread_offline();
	rcu_unregister_thread();
	return NULL;
}
Commit	Line	Data
55c2a7f9 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 <assert.h>
	20	#include <inttypes.h>
	21	#include <signal.h>
	22
	23	#include "sessiond-timer.h"
	24	#include "health-sessiond.h"
	25
55c2a7f9 JD	26	static struct timer_signal_data timer_signal = {
	27	.tid = 0,
	28	.setup_done = 0,
	29	.qs_done = 0,
	30	.lock = PTHREAD_MUTEX_INITIALIZER,
	31	};
	32
	33	/*
	34	* Set custom signal mask to current thread.
	35	*/
	36	static void setmask(sigset_t *mask)
	37	{
	38	int ret;
	39
	40	ret = sigemptyset(mask);
	41	if (ret) {
	42	PERROR("sigemptyset");
	43	}
	44	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_TEARDOWN);
	45	if (ret) {
	46	PERROR("sigaddset teardown");
	47	}
	48	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_EXIT);
	49	if (ret) {
	50	PERROR("sigaddset exit");
	51	}
	52	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_ROTATE_PENDING);
	53	if (ret) {
	54	PERROR("sigaddset switch");
	55	}
06e13874 JD	56	ret = sigaddset(mask, LTTNG_SESSIOND_SIG_ROTATE_TIMER);
	57	if (ret) {
	58	PERROR("sigaddset switch");
	59	}
55c2a7f9 JD	60	}
	61
	62	static
	63	void sessiond_timer_signal_thread_qs(unsigned int signr)
	64	{
	65	sigset_t pending_set;
	66	int ret;
	67
	68	/*
	69	* We need to be the only thread interacting with the thread
	70	* that manages signals for teardown synchronization.
	71	*/
	72	pthread_mutex_lock(&timer_signal.lock);
	73
	74	/* Ensure we don't have any signal queued for this session. */
	75	for (;;) {
	76	ret = sigemptyset(&pending_set);
	77	if (ret == -1) {
	78	PERROR("sigemptyset");
	79	}
	80	ret = sigpending(&pending_set);
	81	if (ret == -1) {
	82	PERROR("sigpending");
	83	}
	84	if (!sigismember(&pending_set, signr)) {
	85	break;
	86	}
	87	caa_cpu_relax();
	88	}
	89
	90	/*
	91	* From this point, no new signal handler will be fired that would try to
	92	* access "session". However, we still need to wait for any currently
	93	* executing handler to complete.
	94	*/
	95	cmm_smp_mb();
	96	CMM_STORE_SHARED(timer_signal.qs_done, 0);
	97	cmm_smp_mb();
	98
	99	/*
	100	* Kill with LTTNG_SESSIOND_SIG_TEARDOWN, so signal management thread wakes
	101	* up.
	102	*/
	103	kill(getpid(), LTTNG_SESSIOND_SIG_TEARDOWN);
	104
	105	while (!CMM_LOAD_SHARED(timer_signal.qs_done)) {
	106	caa_cpu_relax();
	107	}
	108	cmm_smp_mb();
	109
	110	pthread_mutex_unlock(&timer_signal.lock);
	111	}
	112
	113	/*
	114	* Start a timer on a session that will fire at a given interval
	115	* (timer_interval_us) and fire a given signal (signal).
	116	*
	117	* Returns a negative value on error, 0 if a timer was created, and
	118	* a positive value if no timer was created (not an error).
	119	*/
	120	static
	121	int session_timer_start(timer_t timer_id, struct ltt_session session,
	122	unsigned int timer_interval_us, int signal)
	123	{
124	int ret = 0, delete_ret;
125	struct sigevent sev;
126	struct itimerspec its;
127
128	assert(session);
129
130	sev.sigev_notify = SIGEV_SIGNAL;
131	sev.sigev_signo = signal;
132	sev.sigev_value.sival_ptr = session;
133	ret = timer_create(CLOCKID, &sev, timer_id);
134	if (ret == -1) {
135	PERROR("timer_create");
136	goto end;
137	}
138
139	its.it_value.tv_sec = timer_interval_us / 1000000;
140	its.it_value.tv_nsec = (timer_interval_us % 1000000) * 1000;
141	its.it_interval.tv_sec = its.it_value.tv_sec;
142	its.it_interval.tv_nsec = its.it_value.tv_nsec;
143
144	ret = timer_settime(*timer_id, 0, &its, NULL);
145	if (ret == -1) {
146	PERROR("timer_settime");
147	goto error_destroy_timer;
148	}
149	goto end;
150
151	error_destroy_timer:
152	delete_ret = timer_delete(*timer_id);
153	if (delete_ret == -1) {
154	PERROR("timer_delete");
155	}
156
157	end:
158	return ret;
159	}
160
161
162	static
163	int session_timer_stop(timer_t *timer_id, int signal)
164	{
165	int ret = 0;
166
167	ret = timer_delete(*timer_id);
168	if (ret == -1) {
169	PERROR("timer_delete");
170	goto end;
171	}
172
173	sessiond_timer_signal_thread_qs(signal);
174	*timer_id = 0;
175	end:
176	return ret;
177	}
178
06e13874 JD	179	int sessiond_timer_rotate_pending_start(struct ltt_session *session,
06e13874 JD	180	unsigned int interval_us)
55c2a7f9 JD	181	{
	182	int ret;
	183
	184	ret = session_timer_start(&session->rotate_relay_pending_timer,
	185	session, interval_us,
	186	LTTNG_SESSIOND_SIG_ROTATE_PENDING);
06e13874 JD	187	if (ret == 0) {
	188	session->rotate_relay_pending_timer_enabled = true;
	189	}
55c2a7f9 JD	190
	191	return ret;
	192	}
	193
0d2ff20e JD	194	/*
	195	* Stop and delete the channel's live timer.
	196	*/
	197	void sessiond_timer_rotate_pending_stop(struct ltt_session *session)
	198	{
	199	int ret;
	200
	201	assert(session);
	202
	203	ret = session_timer_stop(&session->rotate_relay_pending_timer,
	204	LTTNG_SESSIOND_SIG_ROTATE_PENDING);
	205	if (ret == -1) {
	206	ERR("Failed to stop live timer");
	207	}
	208
06e13874 JD	209	session->rotate_relay_pending_timer_enabled = false;
	210	}
	211
	212	int sessiond_rotate_timer_start(struct ltt_session *session,
	213	unsigned int interval_us)
	214	{
	215	int ret;
	216
	217	ret = session_timer_start(&session->rotate_timer, session, interval_us,
	218	LTTNG_SESSIOND_SIG_ROTATE_TIMER);
	219	if (ret == 0) {
	220	session->rotate_timer_enabled = true;
	221	}
	222
	223	return ret;
	224	}
	225
	226	/*
	227	* Stop and delete the channel's live timer.
	228	*/
	229	void sessiond_rotate_timer_stop(struct ltt_session *session)
	230	{
	231	int ret;
	232
	233	assert(session);
	234
	235	ret = session_timer_stop(&session->rotate_timer,
	236	LTTNG_SESSIOND_SIG_ROTATE_TIMER);
	237	if (ret == -1) {
	238	ERR("Failed to stop live timer");
	239	}
	240
	241	session->rotate_timer_enabled = false;
0d2ff20e JD	242	}
0d2ff20e JD	243
55c2a7f9 JD	244	/*
	245	* Block the RT signals for the entire process. It must be called from the
	246	* sessiond main before creating the threads
	247	*/
	248	int sessiond_timer_signal_init(void)
	249	{
	250	int ret;
	251	sigset_t mask;
	252
	253	/* Block signal for entire process, so only our thread processes it. */
	254	setmask(&mask);
	255	ret = pthread_sigmask(SIG_BLOCK, &mask, NULL);
	256	if (ret) {
	257	errno = ret;
	258	PERROR("pthread_sigmask");
	259	return -1;
	260	}
	261	return 0;
	262	}
	263
09a41363 JD	264	static
	265	void relay_rotation_pending_timer(struct timer_thread_parameters *ctx,
	266	int sig, siginfo_t *si)
	267	{
	268	int ret;
	269	struct ltt_session *session = si->si_value.sival_ptr;
06e13874	270	struct sessiond_rotation_timer timer_data;
09a41363 JD	271	assert(session);
09a41363 JD	272
1b58b7f5 JD	273	/*
	274	* Avoid sending too many requests in case the relay is slower to
	275	* respond than the timer period.
	276	*/
	277	if (session->rotate_pending_relay_check_in_progress \|\|
	278	!session->rotate_pending_relay) {
	279	goto end;
	280	}
	281
	282	session->rotate_pending_relay_check_in_progress = true;
06e13874 JD	283	memset(&timer_data, 0, sizeof(struct sessiond_rotation_timer));
	284	timer_data.session_id = session->id;
	285	timer_data.signal = LTTNG_SESSIOND_SIG_ROTATE_PENDING;
	286	ret = lttng_write(ctx->rotate_timer_pipe, &timer_data,
	287	sizeof(timer_data));
09a41363 JD	288	if (ret < sizeof(session->id)) {
	289	PERROR("wakeup rotate pipe");
	290	}
1b58b7f5 JD	291
	292	end:
	293	return;
09a41363 JD	294	}
09a41363 JD	295
06e13874 JD	296	static
	297	void rotate_timer(struct timer_thread_parameters ctx, int sig, siginfo_t si)
	298	{
	299	int ret;
	300	struct ltt_session *session = si->si_value.sival_ptr;
	301	struct sessiond_rotation_timer timer_data;
	302	assert(session);
	303
	304	/*
	305	* No rate limiting here, so if the timer fires too quickly, there will
	306	* be a backlog of timers queued up and we will try to catch up.
	307	*/
	308	memset(&timer_data, 0, sizeof(struct sessiond_rotation_timer));
	309	timer_data.session_id = session->id;
	310	timer_data.signal = LTTNG_SESSIOND_SIG_ROTATE_TIMER;
	311	ret = lttng_write(ctx->rotate_timer_pipe, &timer_data,
	312	sizeof(timer_data));
	313	if (ret < sizeof(session->id)) {
	314	PERROR("wakeup rotate pipe");
	315	}
	316
	317	return;
	318	}
	319
55c2a7f9 JD	320	/*
	321	* This thread is the sighandler for the timer signals.
	322	*/
	323	void sessiond_timer_thread(void data)
	324	{
	325	int signr;
	326	sigset_t mask;
	327	siginfo_t info;
09a41363	328	struct timer_thread_parameters *ctx = data;
55c2a7f9 JD	329
	330	rcu_register_thread();
	331	rcu_thread_online();
	332
	333	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
	334
	335	health_code_update();
	336
	337	/* Only self thread will receive signal mask. */
	338	setmask(&mask);
	339	CMM_STORE_SHARED(timer_signal.tid, pthread_self());
	340
	341	while (1) {
	342	health_code_update();
	343
	344	health_poll_entry();
	345	signr = sigwaitinfo(&mask, &info);
	346	health_poll_exit();
	347
	348	/*
	349	* NOTE: cascading conditions are used instead of a switch case
	350	* since the use of SIGRTMIN in the definition of the signals'
	351	* values prevents the reduction to an integer constant.
	352	*/
	353	if (signr == -1) {
	354	if (errno != EINTR) {
	355	PERROR("sigwaitinfo");
	356	}
	357	continue;
	358	} else if (signr == LTTNG_SESSIOND_SIG_TEARDOWN) {
0d2ff20e	359	fprintf(stderr, "TEARDOWN\n");
55c2a7f9 JD	360	cmm_smp_mb();
	361	CMM_STORE_SHARED(timer_signal.qs_done, 1);
	362	cmm_smp_mb();
	363	DBG("Signal timer metadata thread teardown");
	364	} else if (signr == LTTNG_SESSIOND_SIG_EXIT) {
0d2ff20e	365	fprintf(stderr, "KILL\n");
55c2a7f9	366	goto end;
0d2ff20e	367	} else if (signr == LTTNG_SESSIOND_SIG_ROTATE_PENDING) {
09a41363 JD	368	fprintf(stderr, "PENDING TIMER\n");
09a41363 JD	369	relay_rotation_pending_timer(ctx, info.si_signo, &info);
06e13874 JD	370	} else if (signr == LTTNG_SESSIOND_SIG_ROTATE_TIMER) {
	371	fprintf(stderr, "ROTATE TIMER\n");
	372	rotate_timer(ctx, info.si_signo, &info);
55c2a7f9 JD	373	} else {
	374	ERR("Unexpected signal %d\n", info.si_signo);
	375	}
	376	}
	377
	378	end:
	379	health_unregister(health_sessiond);
	380	rcu_thread_offline();
	381	rcu_unregister_thread();
	382	return NULL;
	383	}