#include "load-session-thread.h"
#include "notification-thread.h"
#include "notification-thread-commands.h"
+#include "rotation-thread.h"
#include "syscall.h"
#include "agent.h"
#include "ht-cleanup.h"
#include "sessiond-config.h"
+#include "sessiond-timer.h"
static const char *help_msg =
#ifdef LTTNG_EMBED_HELP
/* Set to 1 when a SIGUSR1 signal is received. */
static int recv_child_signal;
+static struct lttng_kernel_tracer_version kernel_tracer_version;
+static struct lttng_kernel_tracer_abi_version kernel_tracer_abi_version;
+
/*
* Consumer daemon specific control data. Every value not initialized here is
* set to 0 by the static definition.
.err_sock = -1,
.cmd_sock = -1,
.channel_monitor_pipe = -1,
+ .channel_rotate_pipe = -1,
.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
.lock = PTHREAD_MUTEX_INITIALIZER,
.cond = PTHREAD_COND_INITIALIZER,
.err_sock = -1,
.cmd_sock = -1,
.channel_monitor_pipe = -1,
+ .channel_rotate_pipe = -1,
.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
.lock = PTHREAD_MUTEX_INITIALIZER,
.cond = PTHREAD_COND_INITIALIZER,
.err_sock = -1,
.cmd_sock = -1,
.channel_monitor_pipe = -1,
+ .channel_rotate_pipe = -1,
.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
.lock = PTHREAD_MUTEX_INITIALIZER,
.cond = PTHREAD_COND_INITIALIZER,
static pthread_t agent_reg_thread;
static pthread_t load_session_thread;
static pthread_t notification_thread;
+static pthread_t rotation_thread;
+static pthread_t timer_thread;
/*
* UST registration command queue. This queue is tied with a futex and uses a N
/* Notification thread handle. */
struct notification_thread_handle *notification_thread_handle;
+/* Rotation thread handle. */
+struct rotation_thread_handle *rotation_thread_handle;
+
/* Global hash tables */
struct lttng_ht *agent_apps_ht_by_sock = NULL;
* NR_LTTNG_SESSIOND_READY must match the number of calls to
* sessiond_notify_ready().
*/
-#define NR_LTTNG_SESSIOND_READY 4
+#define NR_LTTNG_SESSIOND_READY 5
int lttng_sessiond_ready = NR_LTTNG_SESSIOND_READY;
int sessiond_check_thread_quit_pipe(int fd, uint32_t events)
PERROR("UST consumerd64 channel monitor pipe close");
}
}
+ if (kconsumer_data.channel_rotate_pipe >= 0) {
+ ret = close(kconsumer_data.channel_rotate_pipe);
+ if (ret < 0) {
+ PERROR("kernel consumer channel rotate pipe close");
+ }
+ }
+ if (ustconsumer32_data.channel_rotate_pipe >= 0) {
+ ret = close(ustconsumer32_data.channel_rotate_pipe);
+ if (ret < 0) {
+ PERROR("UST consumerd32 channel rotate pipe close");
+ }
+ }
+ if (ustconsumer64_data.channel_rotate_pipe >= 0) {
+ ret = close(ustconsumer64_data.channel_rotate_pipe);
+ if (ret < 0) {
+ PERROR("UST consumerd64 channel rotate pipe close");
+ }
+ }
}
/*
health_code_update();
/*
- * Transfer the write-end of the channel monitoring pipe to the
- * by issuing a SET_CHANNEL_MONITOR_PIPE command.
+ * Transfer the write-end of the channel monitoring and rotate pipe
+ * to the consumer by issuing a SET_CHANNEL_MONITOR_PIPE and
+ * SET_CHANNEL_ROTATE_PIPE commands.
*/
cmd_socket_wrapper = consumer_allocate_socket(&consumer_data->cmd_sock);
if (!cmd_socket_wrapper) {
goto error;
}
+ cmd_socket_wrapper->lock = &consumer_data->lock;
ret = consumer_send_channel_monitor_pipe(cmd_socket_wrapper,
consumer_data->channel_monitor_pipe);
if (ret) {
goto error;
}
+
+ ret = consumer_send_channel_rotate_pipe(cmd_socket_wrapper,
+ consumer_data->channel_rotate_pipe);
+ if (ret) {
+ goto error;
+ }
+
/* Discard the socket wrapper as it is no longer needed. */
consumer_destroy_socket(cmd_socket_wrapper);
cmd_socket_wrapper = NULL;
}
/*
- * This thread manage application communication.
+ * 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_manage_apps(void *data)
{
}
/* Validate kernel version */
- ret = kernel_validate_version(kernel_tracer_fd);
+ ret = kernel_validate_version(kernel_tracer_fd, &kernel_tracer_version,
+ &kernel_tracer_abi_version);
if (ret < 0) {
goto error_version;
}
goto error;
}
- /* Create directory(ies) on local filesystem. */
- if (session->kernel_session->consumer->type == CONSUMER_DST_LOCAL &&
- strlen(session->kernel_session->consumer->dst.trace_path) > 0) {
- ret = run_as_mkdir_recursive(
- session->kernel_session->consumer->dst.trace_path,
- S_IRWXU | S_IRWXG, session->uid, session->gid);
- if (ret < 0) {
- if (errno != EEXIST) {
- ERR("Trace directory creation error");
- goto error;
- }
- }
- }
-
session->kernel_session->uid = session->uid;
session->kernel_session->gid = session->gid;
session->kernel_session->output_traces = session->output_traces;
return ret;
}
+static
+struct rotation_thread_timer_queue *create_rotate_timer_queue(void)
+{
+ struct rotation_thread_timer_queue *queue = NULL;
+
+ queue = zmalloc(sizeof(struct rotation_thread_timer_queue));
+ if (!queue) {
+ PERROR("Failed to allocate timer rotate queue");
+ goto end;
+ }
+
+ queue->event_pipe = lttng_pipe_open(FD_CLOEXEC | O_NONBLOCK);
+ CDS_INIT_LIST_HEAD(&queue->list);
+ pthread_mutex_init(&queue->lock, NULL);
+
+end:
+ return queue;
+}
+
+static
+void destroy_rotate_timer_queue(struct rotation_thread_timer_queue *queue)
+{
+ struct sessiond_rotation_timer *node, *tmp_node;
+
+ if (!queue) {
+ return;
+ }
+
+ lttng_pipe_destroy(queue->event_pipe);
+
+ pthread_mutex_lock(&queue->lock);
+ /* Empty wait queue. */
+ cds_list_for_each_entry_safe(node, tmp_node, &queue->list, head) {
+ cds_list_del(&node->head);
+ free(node);
+ }
+ pthread_mutex_unlock(&queue->lock);
+
+ pthread_mutex_destroy(&queue->lock);
+ free(queue);
+}
+
/*
* main
*/
*ust64_channel_monitor_pipe = NULL,
*kernel_channel_monitor_pipe = NULL;
bool notification_thread_running = false;
+ bool rotation_thread_running = false;
+ bool timer_thread_running = false;
+ struct lttng_pipe *ust32_channel_rotate_pipe = NULL,
+ *ust64_channel_rotate_pipe = NULL,
+ *kernel_channel_rotate_pipe = NULL;
+ struct timer_thread_parameters timer_thread_ctx;
+ /* Queue of rotation jobs populated by the sessiond-timer. */
+ struct rotation_thread_timer_queue *rotation_timer_queue = NULL;
init_kernel_workarounds();
goto exit_set_signal_handler;
}
+ if (sessiond_timer_signal_init()) {
+ retval = -1;
+ goto exit_set_signal_handler;
+ }
+
page_size = sysconf(_SC_PAGESIZE);
if (page_size < 0) {
PERROR("sysconf _SC_PAGESIZE");
retval = -1;
goto exit_init_data;
}
+ kernel_channel_rotate_pipe = lttng_pipe_open(0);
+ if (!kernel_channel_rotate_pipe) {
+ ERR("Failed to create kernel consumer channel rotate pipe");
+ retval = -1;
+ goto exit_init_data;
+ }
+ kconsumer_data.channel_rotate_pipe =
+ lttng_pipe_release_writefd(
+ kernel_channel_rotate_pipe);
+ if (kconsumer_data.channel_rotate_pipe < 0) {
+ retval = -1;
+ goto exit_init_data;
+ }
}
lockfile_fd = create_lockfile();
retval = -1;
goto exit_init_data;
}
+ ust32_channel_rotate_pipe = lttng_pipe_open(0);
+ if (!ust32_channel_rotate_pipe) {
+ ERR("Failed to create 32-bit user space consumer channel rotate pipe");
+ retval = -1;
+ goto exit_init_data;
+ }
+ ustconsumer32_data.channel_rotate_pipe = lttng_pipe_release_writefd(
+ ust32_channel_rotate_pipe);
+ if (ustconsumer32_data.channel_rotate_pipe < 0) {
+ retval = -1;
+ goto exit_init_data;
+ }
+
+ /*
+ * The rotation_timer_queue structure is shared between the sessiond timer
+ * thread and the rotation thread. The main() keeps the ownership and
+ * destroys it when both threads have quit.
+ */
+ rotation_timer_queue = create_rotate_timer_queue();
+ if (!rotation_timer_queue) {
+ retval = -1;
+ goto exit_init_data;
+ }
+ timer_thread_ctx.rotation_timer_queue = rotation_timer_queue;
ust64_channel_monitor_pipe = lttng_pipe_open(0);
if (!ust64_channel_monitor_pipe) {
retval = -1;
goto exit_init_data;
}
+ ust64_channel_rotate_pipe = lttng_pipe_open(0);
+ if (!ust64_channel_rotate_pipe) {
+ ERR("Failed to create 64-bit user space consumer channel rotate pipe");
+ retval = -1;
+ goto exit_init_data;
+ }
+ ustconsumer64_data.channel_rotate_pipe = lttng_pipe_release_writefd(
+ ust64_channel_rotate_pipe);
+ if (ustconsumer64_data.channel_rotate_pipe < 0) {
+ retval = -1;
+ goto exit_init_data;
+ }
/*
* See if daemon already exist.
}
notification_thread_running = true;
+ /* Create timer thread. */
+ ret = pthread_create(&timer_thread, default_pthread_attr(),
+ sessiond_timer_thread, &timer_thread_ctx);
+ if (ret) {
+ errno = ret;
+ PERROR("pthread_create timer");
+ retval = -1;
+ stop_threads();
+ goto exit_notification;
+ }
+ timer_thread_running = true;
+
+ /* rotation_thread_data acquires the pipes' read side. */
+ rotation_thread_handle = rotation_thread_handle_create(
+ ust32_channel_rotate_pipe,
+ ust64_channel_rotate_pipe,
+ kernel_channel_rotate_pipe,
+ thread_quit_pipe[0],
+ rotation_timer_queue);
+ if (!rotation_thread_handle) {
+ retval = -1;
+ ERR("Failed to create rotation thread shared data");
+ stop_threads();
+ goto exit_rotation;
+ }
+
+ /* Create rotation thread. */
+ ret = pthread_create(&rotation_thread, default_pthread_attr(),
+ thread_rotation, rotation_thread_handle);
+ if (ret) {
+ errno = ret;
+ PERROR("pthread_create rotation");
+ retval = -1;
+ stop_threads();
+ goto exit_rotation;
+ }
+ rotation_thread_running = true;
+
/* Create thread to manage the client socket */
ret = pthread_create(&client_thread, default_pthread_attr(),
thread_manage_clients, (void *) NULL);
}
exit_client:
+exit_rotation:
exit_notification:
ret = pthread_join(health_thread, &status);
if (ret) {
notification_thread_handle_destroy(notification_thread_handle);
}
+ if (rotation_thread_handle) {
+ if (rotation_thread_running) {
+ ret = pthread_join(rotation_thread, &status);
+ if (ret) {
+ errno = ret;
+ PERROR("pthread_join rotation thread");
+ retval = -1;
+ }
+ }
+ rotation_thread_handle_destroy(rotation_thread_handle);
+ }
+
+ if (timer_thread_running) {
+ kill(getpid(), LTTNG_SESSIOND_SIG_EXIT);
+ ret = pthread_join(timer_thread, &status);
+ if (ret) {
+ errno = ret;
+ PERROR("pthread_join timer thread");
+ retval = -1;
+ }
+ }
+
+ /*
+ * After the rotation and timer thread have quit, we can safely destroy
+ * the rotation_timer_queue.
+ */
+ destroy_rotate_timer_queue(rotation_timer_queue);
+
rcu_thread_offline();
rcu_unregister_thread();
lttng_pipe_destroy(ust32_channel_monitor_pipe);
lttng_pipe_destroy(ust64_channel_monitor_pipe);
lttng_pipe_destroy(kernel_channel_monitor_pipe);
+ lttng_pipe_destroy(ust32_channel_rotate_pipe);
+ lttng_pipe_destroy(ust64_channel_rotate_pipe);
+ lttng_pipe_destroy(kernel_channel_rotate_pipe);
exit_ht_cleanup:
health_app_destroy(health_sessiond);
projects / lttng-tools.git / blobdiff