static pthread_t kernel_thread;
static pthread_t dispatch_thread;
static pthread_t health_thread;
+static pthread_t ht_cleanup_thread;
/*
* UST registration command queue. This queue is tied with a futex and uses a N
futex_nto1_wake(&ust_cmd_queue.futex);
}
+/*
+ * Close every consumer sockets.
+ */
+static void close_consumer_sockets(void)
+{
+ int ret;
+
+ if (kconsumer_data.err_sock >= 0) {
+ ret = close(kconsumer_data.err_sock);
+ if (ret < 0) {
+ PERROR("kernel consumer err_sock close");
+ }
+ }
+ if (ustconsumer32_data.err_sock >= 0) {
+ ret = close(ustconsumer32_data.err_sock);
+ if (ret < 0) {
+ PERROR("UST consumerd32 err_sock close");
+ }
+ }
+ if (ustconsumer64_data.err_sock >= 0) {
+ ret = close(ustconsumer64_data.err_sock);
+ if (ret < 0) {
+ PERROR("UST consumerd64 err_sock close");
+ }
+ }
+ if (kconsumer_data.cmd_sock >= 0) {
+ ret = close(kconsumer_data.cmd_sock);
+ if (ret < 0) {
+ PERROR("kernel consumer cmd_sock close");
+ }
+ }
+ if (ustconsumer32_data.cmd_sock >= 0) {
+ ret = close(ustconsumer32_data.cmd_sock);
+ if (ret < 0) {
+ PERROR("UST consumerd32 cmd_sock close");
+ }
+ }
+ if (ustconsumer64_data.cmd_sock >= 0) {
+ ret = close(ustconsumer64_data.cmd_sock);
+ if (ret < 0) {
+ PERROR("UST consumerd64 cmd_sock close");
+ }
+ }
+}
+
/*
* Cleanup the daemon
*/
DBG("Cleaning up");
- /* First thing first, stop all threads */
+ /*
+ * Close the thread quit pipe. It has already done its job,
+ * since we are now called.
+ */
utils_close_pipe(thread_quit_pipe);
/*
modprobe_remove_lttng_all();
}
+ close_consumer_sockets();
+
/* <fun> */
DBG("%c[%d;%dm*** assert failed :-) *** ==> %c[%dm%c[%d;%dm"
"Matthew, BEET driven development works!%c[%dm",
if (ret < 0) {
goto error;
}
+ /* Update the stream global counter */
+ ksess->stream_count_global += ret;
/*
* Have we already sent fds to the consumer? If yes, it means
pthread_mutex_lock(socket->lock);
ret = kernel_consumer_send_channel_stream(socket,
- channel, ksess);
+ channel, ksess,
+ session->output_traces ? 1 : 0);
pthread_mutex_unlock(socket->lock);
if (ret < 0) {
rcu_read_unlock();
/*
* This first step of the while is to clean this structure which could free
- * non NULL pointers so zero it before the loop.
+ * non NULL pointers so initialize it before the loop.
*/
- memset(&events, 0, sizeof(events));
+ lttng_poll_init(&events);
if (testpoint(thread_manage_kernel)) {
goto error_testpoint;
if (ret) {
PERROR("close");
}
+ consumer_data->err_sock = -1;
}
if (consumer_data->cmd_sock >= 0) {
ret = close(consumer_data->cmd_sock);
if (ret) {
PERROR("close");
}
+ consumer_data->cmd_sock = -1;
}
if (consumer_data->metadata_sock.fd >= 0) {
ret = close(consumer_data->metadata_sock.fd);
goto error;
}
- /* Set socket timeout for both receiving and ending */
+ /*
+ * Set socket timeout for both receiving and ending.
+ * app_socket_timeout is in seconds, whereas
+ * lttcomm_setsockopt_rcv_timeout and
+ * lttcomm_setsockopt_snd_timeout expect msec as
+ * parameter.
+ */
(void) lttcomm_setsockopt_rcv_timeout(sock,
- app_socket_timeout);
+ app_socket_timeout * 1000);
(void) lttcomm_setsockopt_snd_timeout(sock,
- app_socket_timeout);
+ app_socket_timeout * 1000);
DBG("Apps with sock %d added to poll set", sock);
return ret;
}
+/*
+ * Sanitize the wait queue of the dispatch registration thread meaning removing
+ * invalid nodes from it. This is to avoid memory leaks for the case the UST
+ * notify socket is never received.
+ */
+static void sanitize_wait_queue(struct ust_reg_wait_queue *wait_queue)
+{
+ int ret, nb_fd = 0, i;
+ unsigned int fd_added = 0;
+ struct lttng_poll_event events;
+ struct ust_reg_wait_node *wait_node = NULL, *tmp_wait_node;
+
+ assert(wait_queue);
+
+ lttng_poll_init(&events);
+
+ /* Just skip everything for an empty queue. */
+ if (!wait_queue->count) {
+ goto end;
+ }
+
+ ret = lttng_poll_create(&events, wait_queue->count, LTTNG_CLOEXEC);
+ if (ret < 0) {
+ goto error_create;
+ }
+
+ cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
+ &wait_queue->head, head) {
+ assert(wait_node->app);
+ ret = lttng_poll_add(&events, wait_node->app->sock,
+ LPOLLHUP | LPOLLERR);
+ if (ret < 0) {
+ goto error;
+ }
+
+ fd_added = 1;
+ }
+
+ if (!fd_added) {
+ goto end;
+ }
+
+ /*
+ * Poll but don't block so we can quickly identify the faulty events and
+ * clean them afterwards from the wait queue.
+ */
+ ret = lttng_poll_wait(&events, 0);
+ if (ret < 0) {
+ goto error;
+ }
+ nb_fd = ret;
+
+ for (i = 0; i < nb_fd; i++) {
+ /* Get faulty FD. */
+ uint32_t revents = LTTNG_POLL_GETEV(&events, i);
+ int pollfd = LTTNG_POLL_GETFD(&events, i);
+
+ cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
+ &wait_queue->head, head) {
+ if (pollfd == wait_node->app->sock &&
+ (revents & (LPOLLHUP | LPOLLERR))) {
+ cds_list_del(&wait_node->head);
+ wait_queue->count--;
+ ust_app_destroy(wait_node->app);
+ free(wait_node);
+ break;
+ }
+ }
+ }
+
+ if (nb_fd > 0) {
+ DBG("Wait queue sanitized, %d node were cleaned up", nb_fd);
+ }
+
+end:
+ lttng_poll_clean(&events);
+ return;
+
+error:
+ lttng_poll_clean(&events);
+error_create:
+ ERR("Unable to sanitize wait queue");
+ return;
+}
+
/*
* Dispatch request from the registration threads to the application
* communication thread.
*/
static void *thread_dispatch_ust_registration(void *data)
{
- int ret;
+ int ret, err = -1;
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;
+ struct ust_reg_wait_node *wait_node = NULL, *tmp_wait_node;
+ struct ust_reg_wait_queue wait_queue = {
+ .count = 0,
+ };
+
+ health_register(HEALTH_TYPE_APP_REG_DISPATCH);
+
+ health_code_update();
- CDS_LIST_HEAD(wait_queue);
+ CDS_INIT_LIST_HEAD(&wait_queue.head);
DBG("[thread] Dispatch UST command started");
while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
+ health_code_update();
+
/* Atomically prepare the queue futex */
futex_nto1_prepare(&ust_cmd_queue.futex);
struct ust_app *app = NULL;
ust_cmd = NULL;
+ /*
+ * Make sure we don't have node(s) that have hung up before receiving
+ * the notify socket. This is to clean the list in order to avoid
+ * memory leaks from notify socket that are never seen.
+ */
+ sanitize_wait_queue(&wait_queue);
+
+ health_code_update();
/* Dequeue command for registration */
node = cds_wfq_dequeue_blocking(&ust_cmd_queue.queue);
if (node == NULL) {
wait_node = zmalloc(sizeof(*wait_node));
if (!wait_node) {
PERROR("zmalloc wait_node dispatch");
+ 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(ust_cmd);
goto error;
}
* 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);
+ cds_list_add(&wait_node->head, &wait_queue.head);
+ wait_queue.count++;
free(ust_cmd);
/*
* notify socket if found.
*/
cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
- &wait_queue, head) {
+ &wait_queue.head, head) {
+ health_code_update();
if (wait_node->app->pid == ust_cmd->reg_msg.pid) {
wait_node->app->notify_sock = ust_cmd->sock;
cds_list_del(&wait_node->head);
+ wait_queue.count--;
app = wait_node->app;
free(wait_node);
DBG3("UST app notify socket %d is set", ust_cmd->sock);
break;
}
}
+
+ /*
+ * With no application at this stage the received socket is
+ * basically useless so close it before we free the cmd data
+ * structure for good.
+ */
+ if (!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(ust_cmd);
}
rcu_read_unlock();
session_unlock_list();
- } else {
- /* 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);
}
} while (node != NULL);
+ health_poll_entry();
/* Futex wait on queue. Blocking call on futex() */
futex_nto1_wait(&ust_cmd_queue.futex);
+ health_poll_exit();
}
+ /* Normal exit, no error */
+ err = 0;
error:
/* Clean up wait queue. */
cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
- &wait_queue, head) {
+ &wait_queue.head, head) {
cds_list_del(&wait_node->head);
+ wait_queue.count--;
free(wait_node);
}
DBG("Dispatch thread dying");
+ if (err) {
+ health_error();
+ ERR("Health error occurred in %s", __func__);
+ }
+ health_unregister();
return NULL;
}
lus->uid = session->uid;
lus->gid = session->gid;
+ lus->output_traces = session->output_traces;
+ lus->snapshot_mode = session->snapshot_mode;
session->ust_session = lus;
/* Copy session output to the newly created UST session */
session->kernel_session->uid = session->uid;
session->kernel_session->gid = session->gid;
+ session->kernel_session->output_traces = session->output_traces;
+ session->kernel_session->snapshot_mode = session->snapshot_mode;
return LTTNG_OK;
switch (cmd_ctx->lsm->cmd_type) {
case LTTNG_CREATE_SESSION:
+ case LTTNG_CREATE_SESSION_SNAPSHOT:
case LTTNG_DESTROY_SESSION:
case LTTNG_LIST_SESSIONS:
case LTTNG_LIST_DOMAINS:
case LTTNG_START_TRACE:
case LTTNG_STOP_TRACE:
case LTTNG_DATA_PENDING:
+ case LTTNG_SNAPSHOT_ADD_OUTPUT:
+ case LTTNG_SNAPSHOT_DEL_OUTPUT:
+ case LTTNG_SNAPSHOT_LIST_OUTPUT:
+ case LTTNG_SNAPSHOT_RECORD:
need_domain = 0;
break;
default:
/* Commands that DO NOT need a session. */
switch (cmd_ctx->lsm->cmd_type) {
case LTTNG_CREATE_SESSION:
+ case LTTNG_CREATE_SESSION_SNAPSHOT:
case LTTNG_CALIBRATE:
case LTTNG_LIST_SESSIONS:
case LTTNG_LIST_TRACEPOINTS:
session_lock_list();
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 = LTTNG_ERR_SESS_NOT_FOUND;
- } else {
- /* If no session name specified */
- ret = LTTNG_ERR_SELECT_SESS;
- }
+ ret = LTTNG_ERR_SESS_NOT_FOUND;
goto error;
} else {
/* Acquire lock for the session */
ret = cmd_data_pending(cmd_ctx->session);
break;
}
+ case LTTNG_SNAPSHOT_ADD_OUTPUT:
+ {
+ struct lttcomm_lttng_output_id reply;
+
+ ret = cmd_snapshot_add_output(cmd_ctx->session,
+ &cmd_ctx->lsm->u.snapshot_output.output, &reply.id);
+ if (ret != LTTNG_OK) {
+ goto error;
+ }
+
+ ret = setup_lttng_msg(cmd_ctx, sizeof(reply));
+ if (ret < 0) {
+ goto setup_error;
+ }
+
+ /* Copy output list into message payload */
+ memcpy(cmd_ctx->llm->payload, &reply, sizeof(reply));
+ ret = LTTNG_OK;
+ break;
+ }
+ case LTTNG_SNAPSHOT_DEL_OUTPUT:
+ {
+ ret = cmd_snapshot_del_output(cmd_ctx->session,
+ &cmd_ctx->lsm->u.snapshot_output.output);
+ break;
+ }
+ case LTTNG_SNAPSHOT_LIST_OUTPUT:
+ {
+ ssize_t nb_output;
+ struct lttng_snapshot_output *outputs = NULL;
+
+ nb_output = cmd_snapshot_list_outputs(cmd_ctx->session, &outputs);
+ if (nb_output < 0) {
+ ret = -nb_output;
+ goto error;
+ }
+
+ ret = setup_lttng_msg(cmd_ctx,
+ nb_output * sizeof(struct lttng_snapshot_output));
+ if (ret < 0) {
+ free(outputs);
+ goto setup_error;
+ }
+
+ if (outputs) {
+ /* Copy output list into message payload */
+ memcpy(cmd_ctx->llm->payload, outputs,
+ nb_output * sizeof(struct lttng_snapshot_output));
+ free(outputs);
+ }
+
+ ret = LTTNG_OK;
+ break;
+ }
+ case LTTNG_SNAPSHOT_RECORD:
+ {
+ ret = cmd_snapshot_record(cmd_ctx->session,
+ &cmd_ctx->lsm->u.snapshot_record.output,
+ cmd_ctx->lsm->u.snapshot_record.wait);
+ break;
+ }
+ case LTTNG_CREATE_SESSION_SNAPSHOT:
+ {
+ size_t nb_uri, len;
+ struct lttng_uri *uris = NULL;
+
+ nb_uri = cmd_ctx->lsm->u.uri.size;
+ len = nb_uri * sizeof(struct lttng_uri);
+
+ if (nb_uri > 0) {
+ uris = zmalloc(len);
+ if (uris == NULL) {
+ ret = LTTNG_ERR_FATAL;
+ goto error;
+ }
+
+ /* Receive variable len data */
+ DBG("Waiting for %zu URIs from client ...", nb_uri);
+ ret = lttcomm_recv_unix_sock(sock, uris, len);
+ if (ret <= 0) {
+ DBG("No URIs received from client... continuing");
+ *sock_error = 1;
+ 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 = LTTNG_ERR_SESSION_FAIL;
+ free(uris);
+ goto error;
+ }
+ }
+
+ ret = cmd_create_session_snapshot(cmd_ctx->lsm->session.name, uris,
+ nb_uri, &cmd_ctx->creds);
+ free(uris);
+ break;
+ }
default:
ret = LTTNG_ERR_UND;
break;
rcu_register_thread();
+ /* We might hit an error path before this is created. */
+ lttng_poll_init(&events);
+
/* Create unix socket */
sock = lttcomm_create_unix_sock(health_unix_sock_path);
if (sock < 0) {
case LTTNG_HEALTH_CONSUMER:
reply.ret_code = check_consumer_health();
break;
+ case LTTNG_HEALTH_HT_CLEANUP:
+ reply.ret_code = health_check_state(HEALTH_TYPE_HT_CLEANUP);
+ break;
+ case LTTNG_HEALTH_APP_MANAGE_NOTIFY:
+ reply.ret_code = health_check_state(HEALTH_TYPE_APP_MANAGE_NOTIFY);
+ break;
+ case LTTNG_HEALTH_APP_REG_DISPATCH:
+ reply.ret_code = health_check_state(HEALTH_TYPE_APP_REG_DISPATCH);
+ break;
case LTTNG_HEALTH_ALL:
reply.ret_code =
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();
+ check_consumer_health() &&
+ health_check_state(HEALTH_TYPE_HT_CLEANUP) &&
+ health_check_state(HEALTH_TYPE_APP_MANAGE_NOTIFY) &&
+ health_check_state(HEALTH_TYPE_APP_REG_DISPATCH);
break;
default:
reply.ret_code = LTTNG_ERR_UND;
PERROR("close");
}
}
- if (new_sock >= 0) {
- ret = close(new_sock);
- if (ret) {
- PERROR("close");
- }
- }
lttng_poll_clean(&events);
ret = process_client_msg(cmd_ctx, sock, &sock_error);
rcu_thread_offline();
if (ret < 0) {
- if (sock_error) {
- ret = close(sock);
- if (ret) {
- PERROR("close");
- }
- sock = -1;
+ ret = close(sock);
+ if (ret) {
+ PERROR("close");
}
+ sock = -1;
/*
* TODO: Inform client somehow of the fatal error. At
* this point, ret < 0 means that a zmalloc failed
goto error;
}
+ /*
+ * Set the CLOEXEC flag. Return code is useless because either way, the
+ * show must go on.
+ */
+ ret = utils_set_fd_cloexec(consumer_data->err_sock);
+ if (ret < 0) {
+ PERROR("utils_set_fd_cloexec");
+ /* continue anyway */
+ }
+
/* File permission MUST be 660 */
ret = chmod(consumer_data->err_unix_sock_path,
S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
DBG2("Kernel consumer cmd path: %s",
kconsumer_data.cmd_unix_sock_path);
} else {
- home_path = get_home_dir();
+ home_path = utils_get_home_dir();
if (home_path == NULL) {
/* TODO: Add --socket PATH option */
ERR("Can't get HOME directory for sockets creation.");
}
}
+ /* Setup the thread ht_cleanup communication pipe. */
+ if (utils_create_pipe_cloexec(ht_cleanup_pipe) < 0) {
+ goto exit;
+ }
+
/* Setup the thread apps communication pipe. */
if ((ret = utils_create_pipe_cloexec(apps_cmd_pipe)) < 0) {
goto exit;
write_pidfile();
+ /* Initialize communication library */
+ lttcomm_init();
+
+ /* Create thread to manage the client socket */
+ ret = pthread_create(&ht_cleanup_thread, NULL,
+ thread_ht_cleanup, (void *) NULL);
+ if (ret != 0) {
+ PERROR("pthread_create ht_cleanup");
+ goto exit_ht_cleanup;
+ }
+
/* Create thread to manage the client socket */
ret = pthread_create(&health_thread, NULL,
thread_manage_health, (void *) NULL);
}
exit_health:
+ ret = pthread_join(ht_cleanup_thread, &status);
+ if (ret != 0) {
+ PERROR("pthread_join ht cleanup thread");
+ goto error; /* join error, exit without cleanup */
+ }
+exit_ht_cleanup:
exit:
/*
* cleanup() is called when no other thread is running.
projects / lttng-tools.git / blobdiff