X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fmain.c;h=2929bba0d6f677cfdf813e3c89a5bb8d7bc51921;hp=f1cd1b6407f4ee6f34cd18d976368825c3d5d7e3;hb=bd722d76b035766511f0b329f9bbaa2f4180c4ed;hpb=380e8d6f886dfe6633044abfa77739d1210979a3

diff --git a/src/bin/lttng-sessiond/main.c b/src/bin/lttng-sessiond/main.c
index f1cd1b640..2929bba0d 100644
--- a/src/bin/lttng-sessiond/main.c
+++ b/src/bin/lttng-sessiond/main.c
@@ -390,6 +390,51 @@ static void stop_threads(void)
 	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
  */
@@ -401,7 +446,10 @@ static void cleanup(void)
 
 	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);
 
 	/*
@@ -458,6 +506,8 @@ static void cleanup(void)
 		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",
@@ -628,6 +678,8 @@ static int update_kernel_stream(struct consumer_data *consumer_data, int fd)
 				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
@@ -646,7 +698,8 @@ static int update_kernel_stream(struct consumer_data *consumer_data, int fd)
 
 						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();
@@ -1101,12 +1154,14 @@ error:
 		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);
@@ -1334,25 +1389,116 @@ error:
 	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);
 
@@ -1360,6 +1506,14 @@ static void *thread_dispatch_ust_registration(void *data)
 			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) {
@@ -1408,7 +1562,8 @@ static void *thread_dispatch_ust_registration(void *data)
 				 * 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);
 				/*
@@ -1423,10 +1578,12 @@ static void *thread_dispatch_ust_registration(void *data)
 				 * 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);
@@ -1510,19 +1667,29 @@ static void *thread_dispatch_ust_registration(void *data)
 			}
 		} 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;
 }
 
@@ -2265,6 +2432,8 @@ static int create_ust_session(struct ltt_session *session,
 
 	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 */
@@ -2321,6 +2490,8 @@ static int create_kernel_session(struct ltt_session *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;
 
@@ -2376,12 +2547,17 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
 
 	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:
@@ -2434,6 +2610,7 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
 	/* 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:
@@ -2450,12 +2627,7 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
 		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 */
@@ -3069,6 +3241,106 @@ skip_domain:
 		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;
@@ -3232,6 +3504,9 @@ restart:
 		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) &&
@@ -3240,7 +3515,8 @@ restart:
 				health_check_state(HEALTH_TYPE_KERNEL) &&
 				check_consumer_health() &&
 				health_check_state(HEALTH_TYPE_HT_CLEANUP) &&
-				health_check_state(HEALTH_TYPE_APP_MANAGE_NOTIFY);
+				health_check_state(HEALTH_TYPE_APP_MANAGE_NOTIFY) &&
+				health_check_state(HEALTH_TYPE_APP_REG_DISPATCH);
 			break;
 		default:
 			reply.ret_code = LTTNG_ERR_UND;
@@ -3917,6 +4193,16 @@ static int set_consumer_sockets(struct consumer_data *consumer_data,
 		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);
@@ -4145,7 +4431,7 @@ int main(int argc, char **argv)
 		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.");