[lttng-tools.git] / src / bin / lttng-sessiond / manage-apps.c

/*
 * Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
 *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *               2013 - 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.
 */

#include "manage-apps.h"
#include "testpoint.h"
#include "health-sessiond.h"
#include "utils.h"
#include "thread.h"

struct thread_notifiers {
	struct lttng_pipe *quit_pipe;
	int apps_cmd_pipe_read_fd;
};

static void cleanup_application_management_thread(void *data)
{
	struct thread_notifiers *notifiers = data;

	lttng_pipe_destroy(notifiers->quit_pipe);
	free(notifiers);
}

/*
 * This thread receives application command sockets (FDs) on the
 * apps_cmd_pipe and waits (polls) on them until they are closed
 * or an error occurs.
 *
 * At that point, it flushes the data (tracing and metadata) associated
 * with this application and tears down ust app sessions and other
 * associated data structures through ust_app_unregister().
 *
 * Note that this thread never sends commands to the applications
 * through the command sockets; it merely listens for hang-ups
 * and errors on those sockets and cleans-up as they occur.
 */
static void *thread_application_management(void *data)
{
	int i, ret, pollfd, err = -1;
	ssize_t size_ret;
	uint32_t revents, nb_fd;
	struct lttng_poll_event events;
	struct thread_notifiers *notifiers = data;
	const int quit_pipe_read_fd = lttng_pipe_get_readfd(
			notifiers->quit_pipe);

	DBG("[thread] Manage application started");

	rcu_register_thread();
	rcu_thread_online();

	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_APP_MANAGE);

	if (testpoint(sessiond_thread_manage_apps)) {
		goto error_testpoint;
	}

	health_code_update();

	ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
	if (ret < 0) {
		goto error_poll_create;
	}

	ret = lttng_poll_add(&events, notifiers->apps_cmd_pipe_read_fd,
			LPOLLIN | LPOLLRDHUP);
	if (ret < 0) {
		goto error;
	}

	ret = lttng_poll_add(&events, quit_pipe_read_fd, LPOLLIN | LPOLLERR);
	if (ret < 0) {
		goto error;
	}

	if (testpoint(sessiond_thread_manage_apps_before_loop)) {
		goto error;
	}

	health_code_update();

	while (1) {
		DBG("Apps thread polling");

		/* Inifinite blocking call, waiting for transmission */
	restart:
		health_poll_entry();
		ret = lttng_poll_wait(&events, -1);
		DBG("Apps thread return from poll on %d fds",
				LTTNG_POLL_GETNB(&events));
		health_poll_exit();
		if (ret < 0) {
			/*
			 * Restart interrupted system call.
			 */
			if (errno == EINTR) {
				goto restart;
			}
			goto error;
		}

		nb_fd = ret;

		for (i = 0; i < nb_fd; i++) {
			/* Fetch once the poll data */
			revents = LTTNG_POLL_GETEV(&events, i);
			pollfd = LTTNG_POLL_GETFD(&events, i);

			health_code_update();

			if (pollfd == quit_pipe_read_fd) {
				err = 0;
				goto exit;
			} else if (pollfd == notifiers->apps_cmd_pipe_read_fd) {
				/* Inspect the apps cmd pipe */
				if (revents & LPOLLIN) {
					int sock;

					/* Empty pipe */
					size_ret = lttng_read(
							notifiers->apps_cmd_pipe_read_fd,
							&sock, sizeof(sock));
					if (size_ret < sizeof(sock)) {
						PERROR("read apps cmd pipe");
						goto error;
					}

					health_code_update();

					/*
					 * Since this is a command socket (write then read),
					 * we only monitor the error events of the socket.
					 */
					ret = lttng_poll_add(&events, sock,
							LPOLLERR | LPOLLHUP | LPOLLRDHUP);
					if (ret < 0) {
						goto error;
					}

					DBG("Apps with sock %d added to poll set", sock);
				} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
					ERR("Apps command pipe error");
					goto error;
				} else {
					ERR("Unknown poll events %u for sock %d", revents, pollfd);
					goto error;
				}
			} else {
				/*
				 * At this point, we know that a registered application made
				 * the event at poll_wait.
				 */
				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
					/* Removing from the poll set */
					ret = lttng_poll_del(&events, pollfd);
					if (ret < 0) {
						goto error;
					}

					/* Socket closed on remote end. */
					ust_app_unregister(pollfd);
				} else {
					ERR("Unexpected poll events %u for sock %d", revents, pollfd);
					goto error;
				}
			}

			health_code_update();
		}
	}

exit:
error:
	lttng_poll_clean(&events);
error_poll_create:
error_testpoint:

	/*
	 * We don't clean the UST app hash table here since already registered
	 * applications can still be controlled so let them be until the session
	 * daemon dies or the applications stop.
	 */

	if (err) {
		health_error();
		ERR("Health error occurred in %s", __func__);
	}
	health_unregister(health_sessiond);
	DBG("Application communication apps thread cleanup complete");
	rcu_thread_offline();
	rcu_unregister_thread();
	return NULL;
}

static bool shutdown_application_management_thread(void *data)
{
	struct thread_notifiers *notifiers = data;
	const int write_fd = lttng_pipe_get_writefd(notifiers->quit_pipe);

	return notify_thread_pipe(write_fd) == 1;
}

bool launch_application_management_thread(int apps_cmd_pipe_read_fd)
{
	struct lttng_pipe *quit_pipe;
	struct thread_notifiers *notifiers = NULL;
	struct lttng_thread *thread;

	notifiers = zmalloc(sizeof(*notifiers));
	if (!notifiers) {
		goto error_alloc;
	}
	quit_pipe = lttng_pipe_open(FD_CLOEXEC);
	if (!quit_pipe) {
		goto error;
	}
	notifiers->quit_pipe = quit_pipe;
	notifiers->apps_cmd_pipe_read_fd = apps_cmd_pipe_read_fd;

	thread = lttng_thread_create("UST application management",
			thread_application_management,
			shutdown_application_management_thread,
			cleanup_application_management_thread,
			notifiers);
	if (!thread) {
		goto error;
	}

	lttng_thread_put(thread);
	return true;
error:
	cleanup_application_management_thread(notifiers);
error_alloc:
	return false;
}
Commit	Line	Data
7649924e JG	1	/*
	2	* Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
	3	* Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	4	* 2013 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
	5	*
	6	* This program is free software; you can redistribute it and/or modify
	7	* it under the terms of the GNU General Public License, version 2 only,
	8	* as published by the Free Software Foundation.
	9	*
	10	* This program is distributed in the hope that it will be useful,
	11	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	12	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	13	* GNU General Public License for more details.
	14	*
	15	* You should have received a copy of the GNU General Public License along
	16	* with this program; if not, write to the Free Software Foundation, Inc.,
	17	* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	18	*/
	19
	20	#include "manage-apps.h"
	21	#include "testpoint.h"
	22	#include "health-sessiond.h"
	23	#include "utils.h"
	24	#include "thread.h"
	25
	26	struct thread_notifiers {
	27	struct lttng_pipe *quit_pipe;
	28	int apps_cmd_pipe_read_fd;
	29	};
	30
	31	static void cleanup_application_management_thread(void *data)
	32	{
	33	struct thread_notifiers *notifiers = data;
	34
	35	lttng_pipe_destroy(notifiers->quit_pipe);
	36	free(notifiers);
	37	}
	38
	39	/*
	40	* This thread receives application command sockets (FDs) on the
	41	* apps_cmd_pipe and waits (polls) on them until they are closed
	42	* or an error occurs.
	43	*
	44	* At that point, it flushes the data (tracing and metadata) associated
	45	* with this application and tears down ust app sessions and other
	46	* associated data structures through ust_app_unregister().
	47	*
	48	* Note that this thread never sends commands to the applications
	49	* through the command sockets; it merely listens for hang-ups
	50	* and errors on those sockets and cleans-up as they occur.
	51	*/
	52	static void thread_application_management(void data)
	53	{
	54	int i, ret, pollfd, err = -1;
	55	ssize_t size_ret;
	56	uint32_t revents, nb_fd;
	57	struct lttng_poll_event events;
	58	struct thread_notifiers *notifiers = data;
	59	const int quit_pipe_read_fd = lttng_pipe_get_readfd(
	60	notifiers->quit_pipe);
	61
	62	DBG("[thread] Manage application started");
	63
	64	rcu_register_thread();
65	rcu_thread_online();
66
67	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_APP_MANAGE);
68
69	if (testpoint(sessiond_thread_manage_apps)) {
70	goto error_testpoint;
71	}
72
73	health_code_update();
74
75	ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
76	if (ret < 0) {
77	goto error_poll_create;
78	}
79
80	ret = lttng_poll_add(&events, notifiers->apps_cmd_pipe_read_fd,
81	LPOLLIN \| LPOLLRDHUP);
82	if (ret < 0) {
83	goto error;
84	}
85
86	ret = lttng_poll_add(&events, quit_pipe_read_fd, LPOLLIN \| LPOLLERR);
87	if (ret < 0) {
88	goto error;
89	}
90
91	if (testpoint(sessiond_thread_manage_apps_before_loop)) {
92	goto error;
93	}
94
95	health_code_update();
96
97	while (1) {
98	DBG("Apps thread polling");
99
100	/* Inifinite blocking call, waiting for transmission */
101	restart:
102	health_poll_entry();
103	ret = lttng_poll_wait(&events, -1);
104	DBG("Apps thread return from poll on %d fds",
105	LTTNG_POLL_GETNB(&events));
106	health_poll_exit();
107	if (ret < 0) {
108	/*
109	* Restart interrupted system call.
110	*/
111	if (errno == EINTR) {
112	goto restart;
113	}
114	goto error;
115	}
116
117	nb_fd = ret;
118
119	for (i = 0; i < nb_fd; i++) {
120	/* Fetch once the poll data */
121	revents = LTTNG_POLL_GETEV(&events, i);
122	pollfd = LTTNG_POLL_GETFD(&events, i);
123
124	health_code_update();
125
7649924e JG	126	if (pollfd == quit_pipe_read_fd) {
	127	err = 0;
	128	goto exit;
	129	} else if (pollfd == notifiers->apps_cmd_pipe_read_fd) {
	130	/* Inspect the apps cmd pipe */
	131	if (revents & LPOLLIN) {
	132	int sock;
	133
	134	/* Empty pipe */
	135	size_ret = lttng_read(
	136	notifiers->apps_cmd_pipe_read_fd,
	137	&sock, sizeof(sock));
	138	if (size_ret < sizeof(sock)) {
	139	PERROR("read apps cmd pipe");
	140	goto error;
	141	}
	142
	143	health_code_update();
	144
	145	/*
	146	* Since this is a command socket (write then read),
	147	* we only monitor the error events of the socket.
	148	*/
	149	ret = lttng_poll_add(&events, sock,
	150	LPOLLERR \| LPOLLHUP \| LPOLLRDHUP);
	151	if (ret < 0) {
	152	goto error;
	153	}
	154
	155	DBG("Apps with sock %d added to poll set", sock);
	156	} else if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	157	ERR("Apps command pipe error");
	158	goto error;
	159	} else {
	160	ERR("Unknown poll events %u for sock %d", revents, pollfd);
	161	goto error;
	162	}
	163	} else {
	164	/*
	165	* At this point, we know that a registered application made
	166	* the event at poll_wait.
	167	*/
	168	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	169	/* Removing from the poll set */
	170	ret = lttng_poll_del(&events, pollfd);
	171	if (ret < 0) {
	172	goto error;
	173	}
	174
	175	/* Socket closed on remote end. */
	176	ust_app_unregister(pollfd);
	177	} else {
	178	ERR("Unexpected poll events %u for sock %d", revents, pollfd);
	179	goto error;
	180	}
	181	}
	182
	183	health_code_update();
	184	}
	185	}
	186
	187	exit:
	188	error:
	189	lttng_poll_clean(&events);
190	error_poll_create:
191	error_testpoint:
192
193	/*
194	* We don't clean the UST app hash table here since already registered
195	* applications can still be controlled so let them be until the session
196	* daemon dies or the applications stop.
197	*/
198
199	if (err) {
200	health_error();
201	ERR("Health error occurred in %s", __func__);
202	}
203	health_unregister(health_sessiond);
204	DBG("Application communication apps thread cleanup complete");
205	rcu_thread_offline();
206	rcu_unregister_thread();
207	return NULL;
208	}
209
210	static bool shutdown_application_management_thread(void *data)
211	{
212	struct thread_notifiers *notifiers = data;
213	const int write_fd = lttng_pipe_get_writefd(notifiers->quit_pipe);
214
215	return notify_thread_pipe(write_fd) == 1;
216	}
217
218	bool launch_application_management_thread(int apps_cmd_pipe_read_fd)
219	{
220	struct lttng_pipe *quit_pipe;
221	struct thread_notifiers *notifiers = NULL;
222	struct lttng_thread *thread;
223
7649924e JG	224	notifiers = zmalloc(sizeof(*notifiers));
7649924e JG	225	if (!notifiers) {
21fa020e JG	226	goto error_alloc;
	227	}
	228	quit_pipe = lttng_pipe_open(FD_CLOEXEC);
	229	if (!quit_pipe) {
7649924e JG	230	goto error;
	231	}
	232	notifiers->quit_pipe = quit_pipe;
	233	notifiers->apps_cmd_pipe_read_fd = apps_cmd_pipe_read_fd;
	234
	235	thread = lttng_thread_create("UST application management",
	236	thread_application_management,
	237	shutdown_application_management_thread,
	238	cleanup_application_management_thread,
	239	notifiers);
	240	if (!thread) {
	241	goto error;
	242	}
	243
	244	lttng_thread_put(thread);
	245	return true;
	246	error:
	247	cleanup_application_management_thread(notifiers);
21fa020e	248	error_alloc:
7649924e JG	249	return false;
7649924e JG	250	}