#include <grp.h>
#include <limits.h>
#include <pthread.h>
-#include <semaphore.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include "fd-limit.h"
#include "filter.h"
#include "health.h"
+#include "testpoint.h"
#define CONSUMERD_FILE "lttng-consumerd"
static pid_t ppid; /* Parent PID for --sig-parent option */
static char *rundir;
-/* Consumer daemon specific control data */
+/*
+ * Consumer daemon specific control data. Every value not initialized here is
+ * set to 0 by the static definition.
+ */
static struct consumer_data kconsumer_data = {
.type = LTTNG_CONSUMER_KERNEL,
.err_unix_sock_path = DEFAULT_KCONSUMERD_ERR_SOCK_PATH,
.cmd_sock = -1,
.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
.lock = PTHREAD_MUTEX_INITIALIZER,
+ .cond = PTHREAD_COND_INITIALIZER,
+ .cond_mutex = PTHREAD_MUTEX_INITIALIZER,
};
static struct consumer_data ustconsumer64_data = {
.type = LTTNG_CONSUMER64_UST,
.cmd_sock = -1,
.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
.lock = PTHREAD_MUTEX_INITIALIZER,
+ .cond = PTHREAD_COND_INITIALIZER,
+ .cond_mutex = PTHREAD_MUTEX_INITIALIZER,
};
static struct consumer_data ustconsumer32_data = {
.type = LTTNG_CONSUMER32_UST,
.cmd_sock = -1,
.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
.lock = PTHREAD_MUTEX_INITIALIZER,
+ .cond = PTHREAD_COND_INITIALIZER,
+ .cond_mutex = PTHREAD_MUTEX_INITIALIZER,
};
/* Shared between threads */
DBG("Thread manage kernel started");
+ testpoint(thread_manage_kernel);
+
health_code_update(&health_thread_kernel);
+ testpoint(thread_manage_kernel_before_loop);
+
ret = create_thread_poll_set(&events, 2);
if (ret < 0) {
goto error_poll_create;
return NULL;
}
+/*
+ * Signal pthread condition of the consumer data that the thread.
+ */
+static void signal_consumer_condition(struct consumer_data *data, int state)
+{
+ pthread_mutex_lock(&data->cond_mutex);
+
+ /*
+ * The state is set before signaling. It can be any value, it's the waiter
+ * job to correctly interpret this condition variable associated to the
+ * consumer pthread_cond.
+ *
+ * A value of 0 means that the corresponding thread of the consumer data
+ * was not started. 1 indicates that the thread has started and is ready
+ * for action. A negative value means that there was an error during the
+ * thread bootstrap.
+ */
+ data->consumer_thread_is_ready = state;
+ (void) pthread_cond_signal(&data->cond);
+
+ pthread_mutex_unlock(&data->cond_mutex);
+}
+
/*
* This thread manage the consumer error sent back to the session daemon.
*/
/* Inifinite blocking call, waiting for transmission */
restart:
health_poll_update(&consumer_data->health);
+
+ testpoint(thread_manage_consumer);
+
ret = lttng_poll_wait(&events, -1);
health_poll_update(&consumer_data->health);
if (ret < 0) {
health_code_update(&consumer_data->health);
- if (code == CONSUMERD_COMMAND_SOCK_READY) {
+ if (code == LTTCOMM_CONSUMERD_COMMAND_SOCK_READY) {
consumer_data->cmd_sock =
lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path);
if (consumer_data->cmd_sock < 0) {
- sem_post(&consumer_data->sem);
+ /* On error, signal condition and quit. */
+ signal_consumer_condition(consumer_data, -1);
PERROR("consumer connect");
goto error;
}
- /* Signal condition to tell that the kconsumerd is ready */
- sem_post(&consumer_data->sem);
- DBG("consumer command socket ready");
+ signal_consumer_condition(consumer_data, 1);
+ DBG("Consumer command socket ready");
} else {
ERR("consumer error when waiting for SOCK_READY : %s",
lttcomm_get_readable_code(-code));
DBG("[thread] Manage application started");
+ testpoint(thread_manage_apps);
+
rcu_register_thread();
rcu_thread_online();
goto error;
}
+ testpoint(thread_manage_apps_before_loop);
+
health_code_update(&health_thread_app_manage);
while (1) {
DBG("[thread] Manage application registration started");
+ testpoint(thread_registration_apps);
+
ret = lttcomm_listen_unix_sock(apps_sock);
if (ret < 0) {
goto error_listen;
*/
static int spawn_consumer_thread(struct consumer_data *consumer_data)
{
- int ret;
+ int ret, clock_ret;
struct timespec timeout;
- timeout.tv_sec = DEFAULT_SEM_WAIT_TIMEOUT;
- timeout.tv_nsec = 0;
+ /* Make sure we set the readiness flag to 0 because we are NOT ready */
+ consumer_data->consumer_thread_is_ready = 0;
- /* Setup semaphore */
- ret = sem_init(&consumer_data->sem, 0, 0);
- if (ret < 0) {
- PERROR("sem_init consumer semaphore");
+ /* Setup pthread condition */
+ ret = pthread_condattr_init(&consumer_data->condattr);
+ if (ret != 0) {
+ errno = ret;
+ PERROR("pthread_condattr_init consumer data");
goto error;
}
- ret = pthread_create(&consumer_data->thread, NULL,
- thread_manage_consumer, consumer_data);
+ /*
+ * Set the monotonic clock in order to make sure we DO NOT jump in time
+ * between the clock_gettime() call and the timedwait call. See bug #324
+ * for a more details and how we noticed it.
+ */
+ ret = pthread_condattr_setclock(&consumer_data->condattr, CLOCK_MONOTONIC);
+ if (ret != 0) {
+ errno = ret;
+ PERROR("pthread_condattr_setclock consumer data");
+ goto error;
+ }
+
+ ret = pthread_cond_init(&consumer_data->cond, &consumer_data->condattr);
+ if (ret != 0) {
+ errno = ret;
+ PERROR("pthread_cond_init consumer data");
+ goto error;
+ }
+
+ ret = pthread_create(&consumer_data->thread, NULL, thread_manage_consumer,
+ consumer_data);
if (ret != 0) {
PERROR("pthread_create consumer");
ret = -1;
goto error;
}
+ /* We are about to wait on a pthread condition */
+ pthread_mutex_lock(&consumer_data->cond_mutex);
+
/* Get time for sem_timedwait absolute timeout */
- ret = clock_gettime(CLOCK_REALTIME, &timeout);
- if (ret < 0) {
- PERROR("clock_gettime spawn consumer");
- /* Infinite wait for the kconsumerd thread to be ready */
- ret = sem_wait(&consumer_data->sem);
- } else {
- /* Normal timeout if the gettime was successful */
- timeout.tv_sec += DEFAULT_SEM_WAIT_TIMEOUT;
- ret = sem_timedwait(&consumer_data->sem, &timeout);
+ clock_ret = clock_gettime(CLOCK_MONOTONIC, &timeout);
+ /*
+ * Set the timeout for the condition timed wait even if the clock gettime
+ * call fails since we might loop on that call and we want to avoid to
+ * increment the timeout too many times.
+ */
+ timeout.tv_sec += DEFAULT_SEM_WAIT_TIMEOUT;
+
+ /*
+ * The following loop COULD be skipped in some conditions so this is why we
+ * set ret to 0 in order to make sure at least one round of the loop is
+ * done.
+ */
+ ret = 0;
+
+ /*
+ * Loop until the condition is reached or when a timeout is reached. Note
+ * that the pthread_cond_timedwait(P) man page specifies that EINTR can NOT
+ * be returned but the pthread_cond(3), from the glibc-doc, says that it is
+ * possible. This loop does not take any chances and works with both of
+ * them.
+ */
+ while (!consumer_data->consumer_thread_is_ready && ret != ETIMEDOUT) {
+ if (clock_ret < 0) {
+ PERROR("clock_gettime spawn consumer");
+ /* Infinite wait for the consumerd thread to be ready */
+ ret = pthread_cond_wait(&consumer_data->cond,
+ &consumer_data->cond_mutex);
+ } else {
+ ret = pthread_cond_timedwait(&consumer_data->cond,
+ &consumer_data->cond_mutex, &timeout);
+ }
}
- if (ret < 0) {
- if (errno == ETIMEDOUT) {
+ /* Release the pthread condition */
+ pthread_mutex_unlock(&consumer_data->cond_mutex);
+
+ if (ret != 0) {
+ errno = ret;
+ if (ret == ETIMEDOUT) {
/*
* Call has timed out so we kill the kconsumerd_thread and return
* an error.
*/
- ERR("The consumer thread was never ready. Killing it");
+ ERR("Condition timed out. The consumer thread was never ready."
+ " Killing it");
ret = pthread_cancel(consumer_data->thread);
if (ret < 0) {
PERROR("pthread_cancel consumer thread");
}
} else {
- PERROR("semaphore wait failed consumer thread");
+ PERROR("pthread_cond_wait failed consumer thread");
}
goto error;
}
pthread_mutex_lock(&consumer_data->pid_mutex);
if (consumer_data->pid == 0) {
- ERR("Kconsumerd did not start");
+ ERR("Consumerd did not start");
pthread_mutex_unlock(&consumer_data->pid_mutex);
goto error;
}
PERROR("close");
}
kernel_tracer_fd = -1;
- return LTTCOMM_KERN_VERSION;
+ return LTTNG_ERR_KERN_VERSION;
error_modules:
ret = close(kernel_tracer_fd);
WARN("No kernel tracer available");
kernel_tracer_fd = -1;
if (!is_root) {
- return LTTCOMM_NEED_ROOT_SESSIOND;
+ return LTTNG_ERR_NEED_ROOT_SESSIOND;
} else {
- return LTTCOMM_KERN_NA;
+ return LTTNG_ERR_KERN_NA;
}
}
switch (domain) {
case LTTNG_DOMAIN_KERNEL:
DBG3("Copying tracing session consumer output in kernel session");
+ /*
+ * XXX: We should audit the session creation and what this function
+ * does "extra" in order to avoid a destroy since this function is used
+ * in the domain session creation (kernel and ust) only. Same for UST
+ * domain.
+ */
+ if (session->kernel_session->consumer) {
+ consumer_destroy_output(session->kernel_session->consumer);
+ }
session->kernel_session->consumer =
consumer_copy_output(session->consumer);
/* Ease our life a bit for the next part */
break;
case LTTNG_DOMAIN_UST:
DBG3("Copying tracing session consumer output in UST session");
+ if (session->ust_session->consumer) {
+ consumer_destroy_output(session->ust_session->consumer);
+ }
session->ust_session->consumer =
consumer_copy_output(session->consumer);
/* Ease our life a bit for the next part */
dir_name = DEFAULT_UST_TRACE_DIR;
break;
default:
- ret = LTTCOMM_UNKNOWN_DOMAIN;
+ ret = LTTNG_ERR_UNKNOWN_DOMAIN;
goto error;
}
sizeof(consumer->subdir) - strlen(consumer->subdir) - 1);
DBG3("Copy session consumer subdir %s", consumer->subdir);
- ret = LTTCOMM_OK;
+ ret = LTTNG_OK;
error:
return ret;
break;
default:
ERR("Unknown UST domain on create session %d", domain->type);
- ret = LTTCOMM_UNKNOWN_DOMAIN;
+ ret = LTTNG_ERR_UNKNOWN_DOMAIN;
goto error;
}
lus = trace_ust_create_session(session->path, session->id, domain);
if (lus == NULL) {
- ret = LTTCOMM_UST_SESS_FAIL;
+ ret = LTTNG_ERR_UST_SESS_FAIL;
goto error;
}
/* Copy session output to the newly created UST session */
ret = copy_session_consumer(domain->type, session);
- if (ret != LTTCOMM_OK) {
+ if (ret != LTTNG_OK) {
goto error;
}
- return LTTCOMM_OK;
+ return LTTNG_OK;
error:
free(lus);
ret = kernel_create_session(session, kernel_tracer_fd);
if (ret < 0) {
- ret = LTTCOMM_KERN_SESS_FAIL;
+ ret = LTTNG_ERR_KERN_SESS_FAIL;
goto error;
}
/* Copy session output to the newly created Kernel session */
ret = copy_session_consumer(LTTNG_DOMAIN_KERNEL, session);
- if (ret != LTTCOMM_OK) {
+ if (ret != LTTNG_OK) {
goto error;
}
session->kernel_session->uid = session->uid;
session->kernel_session->gid = session->gid;
- return LTTCOMM_OK;
+ return LTTNG_OK;
error:
trace_kernel_destroy_session(session->kernel_session);
static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
int *sock_error)
{
- int ret = LTTCOMM_OK;
+ int ret = LTTNG_OK;
int need_tracing_session = 1;
int need_domain;
case LTTNG_LIST_DOMAINS:
case LTTNG_START_TRACE:
case LTTNG_STOP_TRACE:
+ case LTTNG_DATA_AVAILABLE:
need_domain = 0;
break;
default:
if (opt_no_kernel && need_domain
&& cmd_ctx->lsm->domain.type == LTTNG_DOMAIN_KERNEL) {
if (!is_root) {
- ret = LTTCOMM_NEED_ROOT_SESSIOND;
+ ret = LTTNG_ERR_NEED_ROOT_SESSIOND;
} else {
- ret = LTTCOMM_KERN_NA;
+ ret = LTTNG_ERR_KERN_NA;
}
goto error;
}
if (cmd_ctx->lsm->cmd_type == LTTNG_REGISTER_CONSUMER) {
pthread_mutex_lock(&kconsumer_data.pid_mutex);
if (kconsumer_data.pid > 0) {
- ret = LTTCOMM_KERN_CONSUMER_FAIL;
+ ret = LTTNG_ERR_KERN_CONSUMER_FAIL;
pthread_mutex_unlock(&kconsumer_data.pid_mutex);
goto error;
}
cmd_ctx->session = session_find_by_name(cmd_ctx->lsm->session.name);
if (cmd_ctx->session == NULL) {
if (cmd_ctx->lsm->session.name != NULL) {
- ret = LTTCOMM_SESS_NOT_FOUND;
+ ret = LTTNG_ERR_SESS_NOT_FOUND;
} else {
/* If no session name specified */
- ret = LTTCOMM_SELECT_SESS;
+ ret = LTTNG_ERR_SELECT_SESS;
}
goto error;
} else {
switch (cmd_ctx->lsm->domain.type) {
case LTTNG_DOMAIN_KERNEL:
if (!is_root) {
- ret = LTTCOMM_NEED_ROOT_SESSIOND;
+ ret = LTTNG_ERR_NEED_ROOT_SESSIOND;
goto error;
}
/* Consumer is in an ERROR state. Report back to client */
if (uatomic_read(&kernel_consumerd_state) == CONSUMER_ERROR) {
- ret = LTTCOMM_NO_KERNCONSUMERD;
+ ret = LTTNG_ERR_NO_KERNCONSUMERD;
goto error;
}
if (cmd_ctx->session->kernel_session == NULL) {
ret = create_kernel_session(cmd_ctx->session);
if (ret < 0) {
- ret = LTTCOMM_KERN_SESS_FAIL;
+ ret = LTTNG_ERR_KERN_SESS_FAIL;
goto error;
}
}
pthread_mutex_unlock(&kconsumer_data.pid_mutex);
ret = start_consumerd(&kconsumer_data);
if (ret < 0) {
- ret = LTTCOMM_KERN_CONSUMER_FAIL;
+ ret = LTTNG_ERR_KERN_CONSUMER_FAIL;
goto error;
}
uatomic_set(&kernel_consumerd_state, CONSUMER_STARTED);
{
/* Consumer is in an ERROR state. Report back to client */
if (uatomic_read(&ust_consumerd_state) == CONSUMER_ERROR) {
- ret = LTTCOMM_NO_USTCONSUMERD;
+ ret = LTTNG_ERR_NO_USTCONSUMERD;
goto error;
}
if (cmd_ctx->session->ust_session == NULL) {
ret = create_ust_session(cmd_ctx->session,
&cmd_ctx->lsm->domain);
- if (ret != LTTCOMM_OK) {
+ if (ret != LTTNG_OK) {
goto error;
}
}
pthread_mutex_unlock(&ustconsumer64_data.pid_mutex);
ret = start_consumerd(&ustconsumer64_data);
if (ret < 0) {
- ret = LTTCOMM_UST_CONSUMER64_FAIL;
+ ret = LTTNG_ERR_UST_CONSUMER64_FAIL;
uatomic_set(&ust_consumerd64_fd, -EINVAL);
goto error;
}
pthread_mutex_unlock(&ustconsumer32_data.pid_mutex);
ret = start_consumerd(&ustconsumer32_data);
if (ret < 0) {
- ret = LTTCOMM_UST_CONSUMER32_FAIL;
+ ret = LTTNG_ERR_UST_CONSUMER32_FAIL;
uatomic_set(&ust_consumerd32_fd, -EINVAL);
goto error;
}
switch (cmd_ctx->lsm->domain.type) {
case LTTNG_DOMAIN_UST:
if (uatomic_read(&ust_consumerd_state) != CONSUMER_STARTED) {
- ret = LTTCOMM_NO_USTCONSUMERD;
+ ret = LTTNG_ERR_NO_USTCONSUMERD;
goto error;
}
break;
case LTTNG_DOMAIN_KERNEL:
if (uatomic_read(&kernel_consumerd_state) != CONSUMER_STARTED) {
- ret = LTTCOMM_NO_KERNCONSUMERD;
+ ret = LTTNG_ERR_NO_KERNCONSUMERD;
goto error;
}
break;
if (!session_access_ok(cmd_ctx->session,
LTTNG_SOCK_GET_UID_CRED(&cmd_ctx->creds),
LTTNG_SOCK_GET_GID_CRED(&cmd_ctx->creds))) {
- ret = LTTCOMM_EPERM;
+ ret = LTTNG_ERR_EPERM;
goto error;
}
}
* be a DOMAIN enuam.
*/
ret = cmd_enable_consumer(cmd_ctx->lsm->domain.type, cmd_ctx->session);
- if (ret != LTTCOMM_OK) {
+ if (ret != LTTNG_OK) {
goto error;
}
nb_events = cmd_list_tracepoints(cmd_ctx->lsm->domain.type, &events);
if (nb_events < 0) {
+ /* Return value is a negative lttng_error_code. */
ret = -nb_events;
goto error;
}
free(events);
- ret = LTTCOMM_OK;
+ ret = LTTNG_OK;
break;
}
case LTTNG_LIST_TRACEPOINT_FIELDS:
nb_fields = cmd_list_tracepoint_fields(cmd_ctx->lsm->domain.type,
&fields);
if (nb_fields < 0) {
+ /* Return value is a negative lttng_error_code. */
ret = -nb_fields;
goto error;
}
free(fields);
- ret = LTTCOMM_OK;
+ ret = LTTNG_OK;
break;
}
case LTTNG_SET_CONSUMER_URI:
len = nb_uri * sizeof(struct lttng_uri);
if (nb_uri == 0) {
- ret = LTTCOMM_INVALID;
+ ret = LTTNG_ERR_INVALID;
goto error;
}
uris = zmalloc(len);
if (uris == NULL) {
- ret = LTTCOMM_FATAL;
+ ret = LTTNG_ERR_FATAL;
goto error;
}
if (ret <= 0) {
DBG("No URIs received from client... continuing");
*sock_error = 1;
- ret = LTTCOMM_SESSION_FAIL;
+ ret = LTTNG_ERR_SESSION_FAIL;
+ free(uris);
goto error;
}
ret = cmd_set_consumer_uri(cmd_ctx->lsm->domain.type, cmd_ctx->session,
nb_uri, uris);
- if (ret != LTTCOMM_OK) {
+ if (ret != LTTNG_OK) {
+ free(uris);
goto error;
}
}
}
+ free(uris);
+
break;
}
case LTTNG_START_TRACE:
if (nb_uri > 0) {
uris = zmalloc(len);
if (uris == NULL) {
- ret = LTTCOMM_FATAL;
+ ret = LTTNG_ERR_FATAL;
goto error;
}
if (ret <= 0) {
DBG("No URIs received from client... continuing");
*sock_error = 1;
- ret = LTTCOMM_SESSION_FAIL;
+ ret = LTTNG_ERR_SESSION_FAIL;
+ free(uris);
goto error;
}
if (nb_uri == 1 && uris[0].dtype != LTTNG_DST_PATH) {
DBG("Creating session with ONE network URI is a bad call");
- ret = LTTCOMM_SESSION_FAIL;
+ ret = LTTNG_ERR_SESSION_FAIL;
+ free(uris);
goto error;
}
}
ret = cmd_create_session_uri(cmd_ctx->lsm->session.name, uris, nb_uri,
&cmd_ctx->creds);
+ free(uris);
+
break;
}
case LTTNG_DESTROY_SESSION:
nb_dom = cmd_list_domains(cmd_ctx->session, &domains);
if (nb_dom < 0) {
+ /* Return value is a negative lttng_error_code. */
ret = -nb_dom;
goto error;
}
free(domains);
- ret = LTTCOMM_OK;
+ ret = LTTNG_OK;
break;
}
case LTTNG_LIST_CHANNELS:
nb_chan = cmd_list_channels(cmd_ctx->lsm->domain.type,
cmd_ctx->session, &channels);
if (nb_chan < 0) {
+ /* Return value is a negative lttng_error_code. */
ret = -nb_chan;
goto error;
}
free(channels);
- ret = LTTCOMM_OK;
+ ret = LTTNG_OK;
break;
}
case LTTNG_LIST_EVENTS:
nb_event = cmd_list_events(cmd_ctx->lsm->domain.type, cmd_ctx->session,
cmd_ctx->lsm->u.list.channel_name, &events);
if (nb_event < 0) {
+ /* Return value is a negative lttng_error_code. */
ret = -nb_event;
goto error;
}
free(events);
- ret = LTTCOMM_OK;
+ ret = LTTNG_OK;
break;
}
case LTTNG_LIST_SESSIONS:
session_unlock_list();
- ret = LTTCOMM_OK;
+ ret = LTTNG_OK;
break;
}
case LTTNG_CALIBRATE:
cdata = &kconsumer_data;
break;
default:
- ret = LTTCOMM_UND;
+ ret = LTTNG_ERR_UND;
goto error;
}
{
struct lttng_filter_bytecode *bytecode;
- if (cmd_ctx->lsm->u.filter.bytecode_len > 65336) {
- ret = LTTCOMM_FILTER_INVAL;
+ if (cmd_ctx->lsm->u.filter.bytecode_len > LTTNG_FILTER_MAX_LEN) {
+ ret = LTTNG_ERR_FILTER_INVAL;
goto error;
}
bytecode = zmalloc(cmd_ctx->lsm->u.filter.bytecode_len);
if (!bytecode) {
- ret = LTTCOMM_FILTER_NOMEM;
+ ret = LTTNG_ERR_FILTER_NOMEM;
goto error;
}
/* Receive var. len. data */
if (ret <= 0) {
DBG("Nothing recv() from client var len data... continuing");
*sock_error = 1;
- ret = LTTCOMM_FILTER_INVAL;
+ ret = LTTNG_ERR_FILTER_INVAL;
goto error;
}
if (bytecode->len + sizeof(*bytecode)
!= cmd_ctx->lsm->u.filter.bytecode_len) {
free(bytecode);
- ret = LTTCOMM_FILTER_INVAL;
+ ret = LTTNG_ERR_FILTER_INVAL;
goto error;
}
bytecode);
break;
}
+ case LTTNG_DATA_AVAILABLE:
+ {
+ ret = cmd_data_available(cmd_ctx->session);
+ break;
+ }
default:
- ret = LTTCOMM_UND;
+ ret = LTTNG_ERR_UND;
break;
}
check_consumer_health();
break;
default:
- reply.ret_code = LTTCOMM_UND;
+ reply.ret_code = LTTNG_ERR_UND;
break;
}
DBG("[thread] Manage client started");
+ testpoint(thread_manage_clients);
+
rcu_register_thread();
health_code_update(&health_thread_cmd);
kill(ppid, SIGUSR1);
}
+ testpoint(thread_manage_clients_before_loop);
+
health_code_update(&health_thread_cmd);
while (1) {
health_init(&ustconsumer64_data.health);
health_poll_update(&ustconsumer64_data.health);
+ /* Not needed anymore. */
+ free(rundir);
+
/* Create thread to manage the client socket */
ret = pthread_create(&health_thread, NULL,
thread_manage_health, (void *) NULL);
projects / lttng-tools.git / blobdiff