#include <config.h>
#include <common/common.h>
-#include <common/compat/poll.h>
#include <common/compat/socket.h>
#include <common/defaults.h>
#include <common/kernel-consumer/kernel-consumer.h>
#include "ust-consumer.h"
#include "utils.h"
#include "fd-limit.h"
-#include "filter.h"
#include "health.h"
#include "testpoint.h"
+#include "ust-thread.h"
#define CONSUMERD_FILE "lttng-consumerd"
/* Const values */
-const char default_home_dir[] = DEFAULT_HOME_DIR;
const char default_tracing_group[] = DEFAULT_TRACING_GROUP;
-const char default_ust_sock_dir[] = DEFAULT_UST_SOCK_DIR;
-const char default_global_apps_pipe[] = DEFAULT_GLOBAL_APPS_PIPE;
const char *progname;
const char *opt_tracing_group;
+static const char *opt_pidfile;
static int opt_sig_parent;
static int opt_verbose_consumer;
static int opt_daemon;
*/
static int apps_cmd_pipe[2] = { -1, -1 };
+int apps_cmd_notify_pipe[2] = { -1, -1 };
+
/* Pthread, Mutexes and Semaphores */
static pthread_t apps_thread;
+static pthread_t apps_notify_thread;
static pthread_t reg_apps_thread;
static pthread_t client_thread;
static pthread_t kernel_thread;
static enum consumerd_state ust_consumerd_state;
static enum consumerd_state kernel_consumerd_state;
-/* Used for the health monitoring of the session daemon. See health.h */
-struct health_state health_thread_cmd;
-struct health_state health_thread_app_manage;
-struct health_state health_thread_app_reg;
-struct health_state health_thread_kernel;
-
/*
* Socket timeout for receiving and sending in seconds.
*/
/*
* Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
*/
-static int create_thread_poll_set(struct lttng_poll_event *events,
- unsigned int size)
+int sessiond_set_thread_pollset(struct lttng_poll_event *events, size_t size)
{
int ret;
- if (events == NULL || size == 0) {
- ret = -1;
- goto error;
- }
+ assert(events);
ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
if (ret < 0) {
}
/* Add quit pipe */
- ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN);
+ ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
if (ret < 0) {
goto error;
}
*
* Return 1 if it was triggered else 0;
*/
-static int check_thread_quit_pipe(int fd, uint32_t events)
+int sessiond_check_thread_quit_pipe(int fd, uint32_t events)
{
if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
return 1;
/* First thing first, stop all threads */
utils_close_pipe(thread_quit_pipe);
+ /*
+ * If opt_pidfile is undefined, the default file will be wiped when
+ * removing the rundir.
+ */
+ if (opt_pidfile) {
+ ret = remove(opt_pidfile);
+ if (ret < 0) {
+ PERROR("remove pidfile %s", opt_pidfile);
+ }
+ }
+
DBG("Removing %s directory", rundir);
ret = asprintf(&cmd, "rm -rf %s", rundir);
if (ret < 0) {
static int send_unix_sock(int sock, void *buf, size_t len)
{
/* Check valid length */
- if (len <= 0) {
+ if (len == 0) {
return -1;
}
struct lttng_ht_iter iter;
struct consumer_socket *socket;
-
+ rcu_read_lock();
cds_lfht_for_each_entry(ksess->consumer->socks->ht,
&iter.iter, socket, node.node) {
/* Code flow error */
assert(socket->fd >= 0);
pthread_mutex_lock(socket->lock);
- ret = kernel_consumer_send_channel_stream(socket->fd,
+ ret = kernel_consumer_send_channel_stream(socket,
channel, ksess);
pthread_mutex_unlock(socket->lock);
if (ret < 0) {
+ rcu_read_unlock();
goto error;
}
}
+ rcu_read_unlock();
}
goto error;
}
}
/*
- * For each tracing session, update newly registered apps.
+ * For each tracing session, update newly registered apps. The session list
+ * lock MUST be acquired before calling this.
*/
static void update_ust_app(int app_sock)
{
struct ltt_session *sess, *stmp;
- session_lock_list();
-
/* For all tracing session(s) */
cds_list_for_each_entry_safe(sess, stmp, &session_list_ptr->head, list) {
session_lock(sess);
}
session_unlock(sess);
}
-
- session_unlock_list();
}
/*
char tmp;
struct lttng_poll_event events;
- DBG("Thread manage kernel started");
-
- testpoint(thread_manage_kernel);
+ DBG("[thread] Thread manage kernel started");
- health_code_update(&health_thread_kernel);
+ health_register(HEALTH_TYPE_KERNEL);
- testpoint(thread_manage_kernel_before_loop);
+ /*
+ * This first step of the while is to clean this structure which could free
+ * non NULL pointers so zero it before the loop.
+ */
+ memset(&events, 0, sizeof(events));
- ret = create_thread_poll_set(&events, 2);
- if (ret < 0) {
- goto error_poll_create;
+ if (testpoint(thread_manage_kernel)) {
+ goto error_testpoint;
}
- ret = lttng_poll_add(&events, kernel_poll_pipe[0], LPOLLIN);
- if (ret < 0) {
- goto error;
+ health_code_update();
+
+ if (testpoint(thread_manage_kernel_before_loop)) {
+ goto error_testpoint;
}
while (1) {
- health_code_update(&health_thread_kernel);
+ health_code_update();
if (update_poll_flag == 1) {
- /*
- * Reset number of fd in the poll set. Always 2 since there is the thread
- * quit pipe and the kernel pipe.
- */
- events.nb_fd = 2;
+ /* Clean events object. We are about to populate it again. */
+ lttng_poll_clean(&events);
+
+ ret = sessiond_set_thread_pollset(&events, 2);
+ if (ret < 0) {
+ goto error_poll_create;
+ }
+ ret = lttng_poll_add(&events, kernel_poll_pipe[0], LPOLLIN);
+ if (ret < 0) {
+ goto error;
+ }
+
+ /* This will add the available kernel channel if any. */
ret = update_kernel_poll(&events);
if (ret < 0) {
goto error;
update_poll_flag = 0;
}
- nb_fd = LTTNG_POLL_GETNB(&events);
-
- DBG("Thread kernel polling on %d fds", nb_fd);
-
- /* Zeroed the poll events */
- lttng_poll_reset(&events);
+ DBG("Thread kernel polling on %d fds", LTTNG_POLL_GETNB(&events));
/* Poll infinite value of time */
restart:
- health_poll_update(&health_thread_kernel);
+ health_poll_entry();
ret = lttng_poll_wait(&events, -1);
- health_poll_update(&health_thread_kernel);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
continue;
}
+ 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(&health_thread_kernel);
+ health_code_update();
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_thread_quit_pipe(pollfd, revents);
+ ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
/* Check for data on kernel pipe */
if (pollfd == kernel_poll_pipe[0] && (revents & LPOLLIN)) {
- ret = read(kernel_poll_pipe[0], &tmp, 1);
+ do {
+ ret = read(kernel_poll_pipe[0], &tmp, 1);
+ } while (ret < 0 && errno == EINTR);
+ /*
+ * Ret value is useless here, if this pipe gets any actions an
+ * update is required anyway.
+ */
update_poll_flag = 1;
continue;
} else {
error:
lttng_poll_clean(&events);
error_poll_create:
+error_testpoint:
utils_close_pipe(kernel_poll_pipe);
kernel_poll_pipe[0] = kernel_poll_pipe[1] = -1;
if (err) {
- health_error(&health_thread_kernel);
+ health_error();
ERR("Health error occurred in %s", __func__);
WARN("Kernel thread died unexpectedly. "
"Kernel tracing can continue but CPU hotplug is disabled.");
}
- health_exit(&health_thread_kernel);
+ health_unregister();
DBG("Kernel thread dying");
return NULL;
}
DBG("[thread] Manage consumer started");
- /*
- * Since the consumer thread can be spawned at any moment in time, we init
- * the health to a poll status (1, which is a valid health over time).
- * When the thread starts, we update here the health to a "code" path being
- * an even value so this thread, when reaching a poll wait, does not
- * trigger an error with an even value.
- *
- * Here is the use case we avoid.
- *
- * +1: the first poll update during initialization (main())
- * +2 * x: multiple code update once in this thread.
- * +1: poll wait in this thread (being a good health state).
- * == even number which after the wait period shows as a bad health.
- *
- * In a nutshell, the following poll update to the health state brings back
- * the state to an even value meaning a code path.
- */
- health_poll_update(&consumer_data->health);
+ health_register(HEALTH_TYPE_CONSUMER);
+
+ health_code_update();
/*
* Pass 2 as size here for the thread quit pipe and kconsumerd_err_sock.
* Nothing more will be added to this poll set.
*/
- ret = create_thread_poll_set(&events, 2);
+ ret = sessiond_set_thread_pollset(&events, 2);
if (ret < 0) {
goto error_poll;
}
goto error;
}
- nb_fd = LTTNG_POLL_GETNB(&events);
-
- health_code_update(&consumer_data->health);
+ health_code_update();
/* Inifinite blocking call, waiting for transmission */
restart:
- health_poll_update(&consumer_data->health);
+ health_poll_entry();
- testpoint(thread_manage_consumer);
+ if (testpoint(thread_manage_consumer)) {
+ goto error;
+ }
ret = lttng_poll_wait(&events, -1);
- health_poll_update(&consumer_data->health);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
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(&consumer_data->health);
+ health_code_update();
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_thread_quit_pipe(pollfd, revents);
+ ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
*/
(void) utils_set_fd_cloexec(sock);
- health_code_update(&consumer_data->health);
+ health_code_update();
DBG2("Receiving code from consumer err_sock");
goto error;
}
- health_code_update(&consumer_data->health);
+ health_code_update();
if (code == LTTCOMM_CONSUMERD_COMMAND_SOCK_READY) {
consumer_data->cmd_sock =
goto error;
}
- health_code_update(&consumer_data->health);
-
- /* Update number of fd */
- nb_fd = LTTNG_POLL_GETNB(&events);
+ health_code_update();
/* Inifinite blocking call, waiting for transmission */
restart_poll:
- health_poll_update(&consumer_data->health);
+ health_poll_entry();
ret = lttng_poll_wait(&events, -1);
- health_poll_update(&consumer_data->health);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
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(&consumer_data->health);
+ health_code_update();
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_thread_quit_pipe(pollfd, revents);
+ ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
}
}
- health_code_update(&consumer_data->health);
+ health_code_update();
/* Wait for any kconsumerd error */
ret = lttcomm_recv_unix_sock(sock, &code,
lttng_poll_clean(&events);
error_poll:
if (err) {
- health_error(&consumer_data->health);
+ health_error();
ERR("Health error occurred in %s", __func__);
}
- health_exit(&consumer_data->health);
+ health_unregister();
DBG("consumer thread cleanup completed");
return NULL;
{
int i, ret, pollfd, err = -1;
uint32_t revents, nb_fd;
- struct ust_command ust_cmd;
struct lttng_poll_event events;
DBG("[thread] Manage application started");
- testpoint(thread_manage_apps);
-
rcu_register_thread();
rcu_thread_online();
- health_code_update(&health_thread_app_manage);
+ health_register(HEALTH_TYPE_APP_MANAGE);
- ret = create_thread_poll_set(&events, 2);
+ if (testpoint(thread_manage_apps)) {
+ goto error_testpoint;
+ }
+
+ health_code_update();
+
+ ret = sessiond_set_thread_pollset(&events, 2);
if (ret < 0) {
goto error_poll_create;
}
goto error;
}
- testpoint(thread_manage_apps_before_loop);
+ if (testpoint(thread_manage_apps_before_loop)) {
+ goto error;
+ }
- health_code_update(&health_thread_app_manage);
+ health_code_update();
while (1) {
- /* Zeroed the events structure */
- lttng_poll_reset(&events);
-
- nb_fd = LTTNG_POLL_GETNB(&events);
-
- DBG("Apps thread polling on %d fds", nb_fd);
+ DBG("Apps thread polling on %d fds", LTTNG_POLL_GETNB(&events));
/* Inifinite blocking call, waiting for transmission */
restart:
- health_poll_update(&health_thread_app_manage);
+ health_poll_entry();
ret = lttng_poll_wait(&events, -1);
- health_poll_update(&health_thread_app_manage);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
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(&health_thread_app_manage);
+ health_code_update();
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_thread_quit_pipe(pollfd, revents);
+ ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
ERR("Apps command pipe error");
goto error;
} else if (revents & LPOLLIN) {
+ int sock;
+
/* Empty pipe */
- ret = read(apps_cmd_pipe[0], &ust_cmd, sizeof(ust_cmd));
- if (ret < 0 || ret < sizeof(ust_cmd)) {
+ do {
+ ret = read(apps_cmd_pipe[0], &sock, sizeof(sock));
+ } while (ret < 0 && errno == EINTR);
+ if (ret < 0 || ret < sizeof(sock)) {
PERROR("read apps cmd pipe");
goto error;
}
- health_code_update(&health_thread_app_manage);
-
- /* Register applicaton to the session daemon */
- ret = ust_app_register(&ust_cmd.reg_msg,
- ust_cmd.sock);
- if (ret == -ENOMEM) {
- goto error;
- } else if (ret < 0) {
- break;
- }
-
- health_code_update(&health_thread_app_manage);
+ health_code_update();
/*
- * Validate UST version compatibility.
+ * We only monitor the error events of the socket. This
+ * thread does not handle any incoming data from UST
+ * (POLLIN).
*/
- ret = ust_app_validate_version(ust_cmd.sock);
- if (ret >= 0) {
- /*
- * Add channel(s) and event(s) to newly registered apps
- * from lttng global UST domain.
- */
- update_ust_app(ust_cmd.sock);
+ ret = lttng_poll_add(&events, sock,
+ LPOLLERR | LPOLLHUP | LPOLLRDHUP);
+ if (ret < 0) {
+ goto error;
}
- health_code_update(&health_thread_app_manage);
+ /* Set socket timeout for both receiving and ending */
+ (void) lttcomm_setsockopt_rcv_timeout(sock,
+ app_socket_timeout);
+ (void) lttcomm_setsockopt_snd_timeout(sock,
+ app_socket_timeout);
- ret = ust_app_register_done(ust_cmd.sock);
- if (ret < 0) {
- /*
- * If the registration is not possible, we simply
- * unregister the apps and continue
- */
- ust_app_unregister(ust_cmd.sock);
- } else {
- /*
- * We only monitor the error events of the socket. This
- * thread does not handle any incoming data from UST
- * (POLLIN).
- */
- ret = lttng_poll_add(&events, ust_cmd.sock,
- LPOLLERR & LPOLLHUP & LPOLLRDHUP);
- if (ret < 0) {
- goto error;
- }
-
- /* Set socket timeout for both receiving and ending */
- (void) lttcomm_setsockopt_rcv_timeout(ust_cmd.sock,
- app_socket_timeout);
- (void) lttcomm_setsockopt_snd_timeout(ust_cmd.sock,
- app_socket_timeout);
+ DBG("Apps with sock %d added to poll set", sock);
- DBG("Apps with sock %d added to poll set",
- ust_cmd.sock);
- }
-
- health_code_update(&health_thread_app_manage);
+ health_code_update();
break;
}
}
}
- health_code_update(&health_thread_app_manage);
+ health_code_update();
}
}
error:
lttng_poll_clean(&events);
error_poll_create:
+error_testpoint:
utils_close_pipe(apps_cmd_pipe);
apps_cmd_pipe[0] = apps_cmd_pipe[1] = -1;
*/
if (err) {
- health_error(&health_thread_app_manage);
+ health_error();
ERR("Health error occurred in %s", __func__);
}
- health_exit(&health_thread_app_manage);
+ health_unregister();
DBG("Application communication apps thread cleanup complete");
rcu_thread_offline();
rcu_unregister_thread();
return NULL;
}
+/*
+ * Send a socket to a thread This is called from the dispatch UST registration
+ * thread once all sockets are set for the application.
+ *
+ * On success, return 0 else a negative value being the errno message of the
+ * write().
+ */
+static int send_socket_to_thread(int fd, int sock)
+{
+ int ret;
+
+ /* Sockets MUST be set or else this should not have been called. */
+ assert(fd >= 0);
+ assert(sock >= 0);
+
+ do {
+ ret = write(fd, &sock, sizeof(sock));
+ } while (ret < 0 && errno == EINTR);
+ if (ret < 0 || ret != sizeof(sock)) {
+ PERROR("write apps pipe %d", fd);
+ if (ret < 0) {
+ ret = -errno;
+ }
+ goto error;
+ }
+
+ /* All good. Don't send back the write positive ret value. */
+ ret = 0;
+error:
+ return ret;
+}
+
/*
* Dispatch request from the registration threads to the application
* communication thread.
int ret;
struct cds_wfq_node *node;
struct ust_command *ust_cmd = NULL;
+ struct {
+ struct ust_app *app;
+ struct cds_list_head head;
+ } *wait_node = NULL, *tmp_wait_node;
+
+ CDS_LIST_HEAD(wait_queue);
DBG("[thread] Dispatch UST command started");
futex_nto1_prepare(&ust_cmd_queue.futex);
do {
+ struct ust_app *app = NULL;
+
/* Dequeue command for registration */
node = cds_wfq_dequeue_blocking(&ust_cmd_queue.queue);
if (node == NULL) {
ust_cmd->reg_msg.uid, ust_cmd->reg_msg.gid,
ust_cmd->sock, ust_cmd->reg_msg.name,
ust_cmd->reg_msg.major, ust_cmd->reg_msg.minor);
- /*
- * Inform apps thread of the new application registration. This
- * call is blocking so we can be assured that the data will be read
- * at some point in time or wait to the end of the world :)
- */
- if (apps_cmd_pipe[1] >= 0) {
- ret = write(apps_cmd_pipe[1], ust_cmd,
- sizeof(struct ust_command));
- if (ret < 0) {
- PERROR("write apps cmd pipe");
- if (errno == EBADF) {
- /*
- * We can't inform the application thread to process
- * registration. We will exit or else application
- * registration will not occur and tracing will never
- * start.
- */
- goto error;
+
+ if (ust_cmd->reg_msg.type == USTCTL_SOCKET_CMD) {
+ wait_node = zmalloc(sizeof(*wait_node));
+ if (!wait_node) {
+ PERROR("zmalloc wait_node dispatch");
+ goto error;
+ }
+ CDS_INIT_LIST_HEAD(&wait_node->head);
+
+ /* Create application object if socket is CMD. */
+ wait_node->app = ust_app_create(&ust_cmd->reg_msg,
+ ust_cmd->sock);
+ if (!wait_node->app) {
+ ret = close(ust_cmd->sock);
+ if (ret < 0) {
+ PERROR("close ust sock dispatch %d", ust_cmd->sock);
+ }
+ lttng_fd_put(1, LTTNG_FD_APPS);
+ free(wait_node);
+ continue;
+ }
+ /*
+ * Add application to the wait queue so we can set the notify
+ * socket before putting this object in the global ht.
+ */
+ cds_list_add(&wait_node->head, &wait_queue);
+
+ /*
+ * We have to continue here since we don't have the notify
+ * socket and the application MUST be added to the hash table
+ * only at that moment.
+ */
+ continue;
+ } else {
+ /*
+ * Look for the application in the local wait queue and set the
+ * notify socket if found.
+ */
+ cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
+ &wait_queue, head) {
+ if (wait_node->app->pid == ust_cmd->reg_msg.pid) {
+ wait_node->app->notify_sock = ust_cmd->sock;
+ cds_list_del(&wait_node->head);
+ app = wait_node->app;
+ free(wait_node);
+ DBG3("UST app notify socket %d is set", ust_cmd->sock);
+ break;
}
}
+ }
+
+ if (app) {
+ /*
+ * @session_lock_list
+ *
+ * Lock the global session list so from the register up to the
+ * registration done message, no thread can see the application
+ * and change its state.
+ */
+ session_lock_list();
+ rcu_read_lock();
+
+ /*
+ * Add application to the global hash table. This needs to be
+ * done before the update to the UST registry can locate the
+ * application.
+ */
+ ust_app_add(app);
+
+ /* Set app version. This call will print an error if needed. */
+ (void) ust_app_version(app);
+
+ /* Send notify socket through the notify pipe. */
+ ret = send_socket_to_thread(apps_cmd_notify_pipe[1],
+ app->notify_sock);
+ if (ret < 0) {
+ rcu_read_unlock();
+ session_unlock_list();
+ /* No notify thread, stop the UST tracing. */
+ goto error;
+ }
+
+ /*
+ * Update newly registered application with the tracing
+ * registry info already enabled information.
+ */
+ update_ust_app(app->sock);
+
+ /*
+ * Don't care about return value. Let the manage apps threads
+ * handle app unregistration upon socket close.
+ */
+ (void) ust_app_register_done(app->sock);
+
+ /*
+ * Even if the application socket has been closed, send the app
+ * to the thread and unregistration will take place at that
+ * place.
+ */
+ ret = send_socket_to_thread(apps_cmd_pipe[1], app->sock);
+ if (ret < 0) {
+ rcu_read_unlock();
+ session_unlock_list();
+ /* No apps. thread, stop the UST tracing. */
+ goto error;
+ }
+
+ rcu_read_unlock();
+ session_unlock_list();
} else {
- /* Application manager thread is not available. */
+ /* Application manager threads are not available. */
ret = close(ust_cmd->sock);
if (ret < 0) {
PERROR("close ust_cmd sock");
}
+ lttng_fd_put(1, LTTNG_FD_APPS);
}
free(ust_cmd);
} while (node != NULL);
}
error:
+ /* Clean up wait queue. */
+ cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
+ &wait_queue, head) {
+ cds_list_del(&wait_node->head);
+ free(wait_node);
+ }
+
DBG("Dispatch thread dying");
return NULL;
}
DBG("[thread] Manage application registration started");
- testpoint(thread_registration_apps);
+ health_register(HEALTH_TYPE_APP_REG);
+
+ if (testpoint(thread_registration_apps)) {
+ goto error_testpoint;
+ }
ret = lttcomm_listen_unix_sock(apps_sock);
if (ret < 0) {
* Pass 2 as size here for the thread quit pipe and apps socket. Nothing
* more will be added to this poll set.
*/
- ret = create_thread_poll_set(&events, 2);
+ ret = sessiond_set_thread_pollset(&events, 2);
if (ret < 0) {
goto error_create_poll;
}
while (1) {
DBG("Accepting application registration");
- nb_fd = LTTNG_POLL_GETNB(&events);
-
/* Inifinite blocking call, waiting for transmission */
restart:
- health_poll_update(&health_thread_app_reg);
+ health_poll_entry();
ret = lttng_poll_wait(&events, -1);
- health_poll_update(&health_thread_app_reg);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
goto error;
}
+ nb_fd = ret;
+
for (i = 0; i < nb_fd; i++) {
- health_code_update(&health_thread_app_reg);
+ health_code_update();
/* Fetch once the poll data */
revents = LTTNG_POLL_GETEV(&events, i);
pollfd = LTTNG_POLL_GETFD(&events, i);
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_thread_quit_pipe(pollfd, revents);
+ ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
sock = -1;
continue;
}
- health_code_update(&health_thread_app_reg);
- ret = lttcomm_recv_unix_sock(sock, &ust_cmd->reg_msg,
- sizeof(struct ust_register_msg));
- if (ret < 0 || ret < sizeof(struct ust_register_msg)) {
- if (ret < 0) {
- PERROR("lttcomm_recv_unix_sock register apps");
- } else {
- ERR("Wrong size received on apps register");
- }
+
+ health_code_update();
+ ret = ust_app_recv_registration(sock, &ust_cmd->reg_msg);
+ if (ret < 0) {
free(ust_cmd);
+ /* Close socket of the application. */
ret = close(sock);
if (ret) {
PERROR("close");
sock = -1;
continue;
}
- health_code_update(&health_thread_app_reg);
+ health_code_update();
ust_cmd->sock = sock;
sock = -1;
exit:
error:
if (err) {
- health_error(&health_thread_app_reg);
+ health_error();
ERR("Health error occurred in %s", __func__);
}
- health_exit(&health_thread_app_reg);
/* Notify that the registration thread is gone */
notify_ust_apps(0);
lttng_poll_clean(&events);
error_listen:
error_create_poll:
+error_testpoint:
DBG("UST Registration thread cleanup complete");
+ health_unregister();
return NULL;
}
static int join_consumer_thread(struct consumer_data *consumer_data)
{
void *status;
- int ret;
/* Consumer pid must be a real one. */
if (consumer_data->pid > 0) {
+ int ret;
ret = kill(consumer_data->pid, SIGTERM);
if (ret) {
ERR("Error killing consumer daemon");
*/
static int start_consumerd(struct consumer_data *consumer_data)
{
- int ret, err;
+ int ret;
/*
* Set the listen() state on the socket since there is a possible race
error:
/* Cleanup already created socket on error. */
if (consumer_data->err_sock >= 0) {
+ int err;
+
err = close(consumer_data->err_sock);
if (err < 0) {
PERROR("close consumer data error socket");
{
int ret;
- ret = health_check_state(&kconsumer_data.health) &&
- health_check_state(&ustconsumer32_data.health) &&
- health_check_state(&ustconsumer64_data.health);
+ ret = health_check_state(HEALTH_TYPE_CONSUMER);
DBG3("Health consumer check %d", ret);
DBG("Creating UST session");
- lus = trace_ust_create_session(session->path, session->id, domain);
+ lus = trace_ust_create_session(session->path, session->id);
if (lus == NULL) {
ret = LTTNG_ERR_UST_SESS_FAIL;
goto error;
/* Start the kernel consumer daemon */
pthread_mutex_lock(&kconsumer_data.pid_mutex);
if (kconsumer_data.pid == 0 &&
- cmd_ctx->lsm->cmd_type != LTTNG_REGISTER_CONSUMER &&
- cmd_ctx->session->start_consumer) {
+ cmd_ctx->lsm->cmd_type != LTTNG_REGISTER_CONSUMER) {
pthread_mutex_unlock(&kconsumer_data.pid_mutex);
ret = start_consumerd(&kconsumer_data);
if (ret < 0) {
pthread_mutex_lock(&ustconsumer64_data.pid_mutex);
if (consumerd64_bin[0] != '\0' &&
ustconsumer64_data.pid == 0 &&
- cmd_ctx->lsm->cmd_type != LTTNG_REGISTER_CONSUMER &&
- cmd_ctx->session->start_consumer) {
+ cmd_ctx->lsm->cmd_type != LTTNG_REGISTER_CONSUMER) {
pthread_mutex_unlock(&ustconsumer64_data.pid_mutex);
ret = start_consumerd(&ustconsumer64_data);
if (ret < 0) {
/* 32-bit */
if (consumerd32_bin[0] != '\0' &&
ustconsumer32_data.pid == 0 &&
- cmd_ctx->lsm->cmd_type != LTTNG_REGISTER_CONSUMER &&
- cmd_ctx->session->start_consumer) {
+ cmd_ctx->lsm->cmd_type != LTTNG_REGISTER_CONSUMER) {
pthread_mutex_unlock(&ustconsumer32_data.pid_mutex);
ret = start_consumerd(&ustconsumer32_data);
if (ret < 0) {
}
}
+ /*
+ * Send relayd information to consumer as soon as we have a domain and a
+ * session defined.
+ */
+ if (cmd_ctx->session && need_domain) {
+ /*
+ * Setup relayd if not done yet. If the relayd information was already
+ * sent to the consumer, this call will gracefully return.
+ */
+ ret = cmd_setup_relayd(cmd_ctx->session);
+ if (ret != LTTNG_OK) {
+ goto error;
+ }
+ }
+
/* Process by command type */
switch (cmd_ctx->lsm->cmd_type) {
case LTTNG_ADD_CONTEXT:
cmd_ctx->lsm->u.disable.channel_name);
break;
}
- case LTTNG_DISABLE_CONSUMER:
- {
- ret = cmd_disable_consumer(cmd_ctx->lsm->domain.type, cmd_ctx->session);
- break;
- }
case LTTNG_ENABLE_CHANNEL:
{
ret = cmd_enable_channel(cmd_ctx->session, cmd_ctx->lsm->domain.type,
&cmd_ctx->lsm->u.channel.chan, kernel_poll_pipe[1]);
break;
}
- case LTTNG_ENABLE_CONSUMER:
- {
- /*
- * XXX: 0 means that this URI should be applied on the session. Should
- * be a DOMAIN enuam.
- */
- ret = cmd_enable_consumer(cmd_ctx->lsm->domain.type, cmd_ctx->session);
- if (ret != LTTNG_OK) {
- goto error;
- }
-
- if (cmd_ctx->lsm->domain.type == 0) {
- /* Add the URI for the UST session if a consumer is present. */
- if (cmd_ctx->session->ust_session &&
- cmd_ctx->session->ust_session->consumer) {
- ret = cmd_enable_consumer(LTTNG_DOMAIN_UST, cmd_ctx->session);
- } else if (cmd_ctx->session->kernel_session &&
- cmd_ctx->session->kernel_session->consumer) {
- ret = cmd_enable_consumer(LTTNG_DOMAIN_KERNEL,
- cmd_ctx->session);
- }
- }
- break;
- }
case LTTNG_ENABLE_EVENT:
{
ret = cmd_enable_event(cmd_ctx->session, cmd_ctx->lsm->domain.type,
cmd_ctx->lsm->u.enable.channel_name,
- &cmd_ctx->lsm->u.enable.event, kernel_poll_pipe[1]);
+ &cmd_ctx->lsm->u.enable.event, NULL, kernel_poll_pipe[1]);
break;
}
case LTTNG_ENABLE_ALL_EVENT:
ret = cmd_enable_event_all(cmd_ctx->session, cmd_ctx->lsm->domain.type,
cmd_ctx->lsm->u.enable.channel_name,
- cmd_ctx->lsm->u.enable.event.type, kernel_poll_pipe[1]);
+ cmd_ctx->lsm->u.enable.event.type, NULL, kernel_poll_pipe[1]);
break;
}
case LTTNG_LIST_TRACEPOINTS:
cmd_ctx->lsm->u.reg.path, cdata);
break;
}
- case LTTNG_SET_FILTER:
+ case LTTNG_ENABLE_EVENT_WITH_FILTER:
{
struct lttng_filter_bytecode *bytecode;
- if (cmd_ctx->lsm->u.filter.bytecode_len > LTTNG_FILTER_MAX_LEN) {
+ if (cmd_ctx->lsm->u.enable.bytecode_len > LTTNG_FILTER_MAX_LEN) {
ret = LTTNG_ERR_FILTER_INVAL;
goto error;
}
- bytecode = zmalloc(cmd_ctx->lsm->u.filter.bytecode_len);
+ if (cmd_ctx->lsm->u.enable.bytecode_len == 0) {
+ ret = LTTNG_ERR_FILTER_INVAL;
+ goto error;
+ }
+ bytecode = zmalloc(cmd_ctx->lsm->u.enable.bytecode_len);
if (!bytecode) {
ret = LTTNG_ERR_FILTER_NOMEM;
goto error;
/* Receive var. len. data */
DBG("Receiving var len data from client ...");
ret = lttcomm_recv_unix_sock(sock, bytecode,
- cmd_ctx->lsm->u.filter.bytecode_len);
+ cmd_ctx->lsm->u.enable.bytecode_len);
if (ret <= 0) {
DBG("Nothing recv() from client var len data... continuing");
*sock_error = 1;
}
if (bytecode->len + sizeof(*bytecode)
- != cmd_ctx->lsm->u.filter.bytecode_len) {
+ != cmd_ctx->lsm->u.enable.bytecode_len) {
free(bytecode);
ret = LTTNG_ERR_FILTER_INVAL;
goto error;
}
- ret = cmd_set_filter(cmd_ctx->session, cmd_ctx->lsm->domain.type,
- cmd_ctx->lsm->u.filter.channel_name,
- cmd_ctx->lsm->u.filter.event_name,
- bytecode);
+ ret = cmd_enable_event(cmd_ctx->session, cmd_ctx->lsm->domain.type,
+ cmd_ctx->lsm->u.enable.channel_name,
+ &cmd_ctx->lsm->u.enable.event, bytecode, kernel_poll_pipe[1]);
break;
}
case LTTNG_DATA_PENDING:
* Pass 2 as size here for the thread quit pipe and client_sock. Nothing
* more will be added to this poll set.
*/
- ret = create_thread_poll_set(&events, 2);
+ ret = sessiond_set_thread_pollset(&events, 2);
if (ret < 0) {
goto error;
}
while (1) {
DBG("Health check ready");
- nb_fd = LTTNG_POLL_GETNB(&events);
-
/* Inifinite blocking call, waiting for transmission */
restart:
ret = lttng_poll_wait(&events, -1);
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);
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_thread_quit_pipe(pollfd, revents);
+ ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
switch (msg.component) {
case LTTNG_HEALTH_CMD:
- reply.ret_code = health_check_state(&health_thread_cmd);
+ reply.ret_code = health_check_state(HEALTH_TYPE_CMD);
break;
case LTTNG_HEALTH_APP_MANAGE:
- reply.ret_code = health_check_state(&health_thread_app_manage);
+ reply.ret_code = health_check_state(HEALTH_TYPE_APP_MANAGE);
break;
case LTTNG_HEALTH_APP_REG:
- reply.ret_code = health_check_state(&health_thread_app_reg);
+ reply.ret_code = health_check_state(HEALTH_TYPE_APP_REG);
break;
case LTTNG_HEALTH_KERNEL:
- reply.ret_code = health_check_state(&health_thread_kernel);
+ reply.ret_code = health_check_state(HEALTH_TYPE_KERNEL);
break;
case LTTNG_HEALTH_CONSUMER:
reply.ret_code = check_consumer_health();
break;
case LTTNG_HEALTH_ALL:
reply.ret_code =
- health_check_state(&health_thread_app_manage) &&
- health_check_state(&health_thread_app_reg) &&
- health_check_state(&health_thread_cmd) &&
- health_check_state(&health_thread_kernel) &&
+ health_check_state(HEALTH_TYPE_APP_MANAGE) &&
+ health_check_state(HEALTH_TYPE_APP_REG) &&
+ health_check_state(HEALTH_TYPE_CMD) &&
+ health_check_state(HEALTH_TYPE_KERNEL) &&
check_consumer_health();
break;
default:
DBG("[thread] Manage client started");
- testpoint(thread_manage_clients);
-
rcu_register_thread();
- health_code_update(&health_thread_cmd);
+ health_register(HEALTH_TYPE_CMD);
+
+ if (testpoint(thread_manage_clients)) {
+ goto error_testpoint;
+ }
+
+ health_code_update();
ret = lttcomm_listen_unix_sock(client_sock);
if (ret < 0) {
* Pass 2 as size here for the thread quit pipe and client_sock. Nothing
* more will be added to this poll set.
*/
- ret = create_thread_poll_set(&events, 2);
+ ret = sessiond_set_thread_pollset(&events, 2);
if (ret < 0) {
goto error_create_poll;
}
kill(ppid, SIGUSR1);
}
- testpoint(thread_manage_clients_before_loop);
+ if (testpoint(thread_manage_clients_before_loop)) {
+ goto error;
+ }
- health_code_update(&health_thread_cmd);
+ health_code_update();
while (1) {
DBG("Accepting client command ...");
- nb_fd = LTTNG_POLL_GETNB(&events);
-
/* Inifinite blocking call, waiting for transmission */
restart:
- health_poll_update(&health_thread_cmd);
+ health_poll_entry();
ret = lttng_poll_wait(&events, -1);
- health_poll_update(&health_thread_cmd);
+ health_poll_exit();
if (ret < 0) {
/*
* Restart interrupted system call.
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(&health_thread_cmd);
+ health_code_update();
/* Thread quit pipe has been closed. Killing thread. */
- ret = check_thread_quit_pipe(pollfd, revents);
+ ret = sessiond_check_thread_quit_pipe(pollfd, revents);
if (ret) {
err = 0;
goto exit;
DBG("Wait for client response");
- health_code_update(&health_thread_cmd);
+ health_code_update();
sock = lttcomm_accept_unix_sock(client_sock);
if (sock < 0) {
cmd_ctx->llm = NULL;
cmd_ctx->session = NULL;
- health_code_update(&health_thread_cmd);
+ health_code_update();
/*
* Data is received from the lttng client. The struct
continue;
}
- health_code_update(&health_thread_cmd);
+ health_code_update();
// TODO: Validate cmd_ctx including sanity check for
// security purpose.
continue;
}
- health_code_update(&health_thread_cmd);
+ health_code_update();
DBG("Sending response (size: %d, retcode: %s)",
cmd_ctx->lttng_msg_size,
clean_command_ctx(&cmd_ctx);
- health_code_update(&health_thread_cmd);
+ health_code_update();
}
exit:
error_listen:
error_create_poll:
+error_testpoint:
unlink(client_unix_sock_path);
if (client_sock >= 0) {
ret = close(client_sock);
}
if (err) {
- health_error(&health_thread_cmd);
+ health_error();
ERR("Health error occurred in %s", __func__);
}
- health_exit(&health_thread_cmd);
+ health_unregister();
DBG("Client thread dying");
fprintf(stderr, " -S, --sig-parent Send SIGCHLD to parent pid to notify readiness.\n");
fprintf(stderr, " -q, --quiet No output at all.\n");
fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n");
+ fprintf(stderr, " -p, --pidfile FILE Write a pid to FILE name overriding the default value.\n");
fprintf(stderr, " --verbose-consumer Verbose mode for consumer. Activate DBG() macro.\n");
fprintf(stderr, " --no-kernel Disable kernel tracer\n");
}
{ "verbose", 0, 0, 'v' },
{ "verbose-consumer", 0, 0, 'Z' },
{ "no-kernel", 0, 0, 'N' },
+ { "pidfile", 1, 0, 'p' },
{ NULL, 0, 0, 0 }
};
while (1) {
int option_index = 0;
- c = getopt_long(argc, argv, "dhqvVSN" "a:c:g:s:C:E:D:F:Z:u:t",
+ c = getopt_long(argc, argv, "dhqvVSN" "a:c:g:s:C:E:D:F:Z:u:t:p:",
long_options, &option_index);
if (c == -1) {
break;
case 'T':
consumerd64_libdir = optarg;
break;
+ case 'p':
+ opt_pidfile = optarg;
+ break;
default:
/* Unknown option or other error.
* Error is printed by getopt, just return */
}
}
+/*
+ * Write pidfile using the rundir and opt_pidfile.
+ */
+static void write_pidfile(void)
+{
+ int ret;
+ char pidfile_path[PATH_MAX];
+
+ assert(rundir);
+
+ if (opt_pidfile) {
+ strncpy(pidfile_path, opt_pidfile, sizeof(pidfile_path));
+ } else {
+ /* Build pidfile path from rundir and opt_pidfile. */
+ ret = snprintf(pidfile_path, sizeof(pidfile_path), "%s/"
+ DEFAULT_LTTNG_SESSIOND_PIDFILE, rundir);
+ if (ret < 0) {
+ PERROR("snprintf pidfile path");
+ goto error;
+ }
+ }
+
+ /*
+ * Create pid file in rundir. Return value is of no importance. The
+ * execution will continue even though we are not able to write the file.
+ */
+ (void) utils_create_pid_file(getpid(), pidfile_path);
+
+error:
+ return;
+}
+
/*
* main
*/
/* Parse arguments */
progname = argv[0];
- if ((ret = parse_args(argc, argv) < 0)) {
+ if ((ret = parse_args(argc, argv)) < 0) {
goto error;
}
/* Set global SHM for ust */
if (strlen(wait_shm_path) == 0) {
snprintf(wait_shm_path, PATH_MAX,
- DEFAULT_HOME_APPS_WAIT_SHM_PATH, geteuid());
+ DEFAULT_HOME_APPS_WAIT_SHM_PATH, getuid());
}
/* Set health check Unix path */
DBG("Client socket path %s", client_unix_sock_path);
DBG("Application socket path %s", apps_unix_sock_path);
+ DBG("Application wait path %s", wait_shm_path);
DBG("LTTng run directory path: %s", rundir);
/* 32 bits consumerd path setup */
goto exit;
}
+ /* Setup the thread apps notify communication pipe. */
+ if (utils_create_pipe_cloexec(apps_cmd_notify_pipe) < 0) {
+ goto exit;
+ }
+
/* Init UST command queue. */
cds_wfq_init(&ust_cmd_queue.queue);
cmd_init();
- /* Init all health thread counters. */
- health_init(&health_thread_cmd);
- health_init(&health_thread_kernel);
- health_init(&health_thread_app_manage);
- health_init(&health_thread_app_reg);
-
- /*
- * Init health counters of the consumer thread. We do a quick hack here to
- * the state of the consumer health is fine even if the thread is not
- * started. Once the thread starts, the health state is updated with a poll
- * value to set a health code path. This is simply to ease our life and has
- * no cost what so ever.
- */
- health_init(&kconsumer_data.health);
- health_poll_update(&kconsumer_data.health);
- health_init(&ustconsumer32_data.health);
- health_poll_update(&ustconsumer32_data.health);
- health_init(&ustconsumer64_data.health);
- health_poll_update(&ustconsumer64_data.health);
-
/* Check for the application socket timeout env variable. */
env_app_timeout = getenv(DEFAULT_APP_SOCKET_TIMEOUT_ENV);
if (env_app_timeout) {
app_socket_timeout = DEFAULT_APP_SOCKET_RW_TIMEOUT;
}
+ write_pidfile();
+
/* Create thread to manage the client socket */
ret = pthread_create(&health_thread, NULL,
thread_manage_health, (void *) NULL);
goto exit_apps;
}
+ /* Create thread to manage application notify socket */
+ ret = pthread_create(&apps_notify_thread, NULL,
+ ust_thread_manage_notify, (void *) NULL);
+ if (ret != 0) {
+ PERROR("pthread_create apps");
+ goto exit_apps;
+ }
+
/* Don't start this thread if kernel tracing is not requested nor root */
if (is_root && !opt_no_kernel) {
/* Create kernel thread to manage kernel event */