X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fmain.c;h=3df9216b035df777e7b085bac81e957754923d02;hb=9563b0adcfa7d982efe7a1ff535a70d2961e3cf7;hp=69bc3a183a9addb3c45b50915a331d743ec13be4;hpb=360e82e5c9b78f970c5318b5862f1646bc9cd882;p=lttng-tools.git

diff --git a/src/bin/lttng-sessiond/main.c b/src/bin/lttng-sessiond/main.c
index 69bc3a183..3df9216b0 100644
--- a/src/bin/lttng-sessiond/main.c
+++ b/src/bin/lttng-sessiond/main.c
@@ -90,7 +90,6 @@ static struct consumer_data kconsumer_data = {
 	.cmd_unix_sock_path = DEFAULT_KCONSUMERD_CMD_SOCK_PATH,
 	.err_sock = -1,
 	.cmd_sock = -1,
-	.metadata_sock.fd = -1,
 	.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
 	.lock = PTHREAD_MUTEX_INITIALIZER,
 	.cond = PTHREAD_COND_INITIALIZER,
@@ -102,7 +101,6 @@ static struct consumer_data ustconsumer64_data = {
 	.cmd_unix_sock_path = DEFAULT_USTCONSUMERD64_CMD_SOCK_PATH,
 	.err_sock = -1,
 	.cmd_sock = -1,
-	.metadata_sock.fd = -1,
 	.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
 	.lock = PTHREAD_MUTEX_INITIALIZER,
 	.cond = PTHREAD_COND_INITIALIZER,
@@ -114,7 +112,6 @@ static struct consumer_data ustconsumer32_data = {
 	.cmd_unix_sock_path = DEFAULT_USTCONSUMERD32_CMD_SOCK_PATH,
 	.err_sock = -1,
 	.cmd_sock = -1,
-	.metadata_sock.fd = -1,
 	.pid_mutex = PTHREAD_MUTEX_INITIALIZER,
 	.lock = PTHREAD_MUTEX_INITIALIZER,
 	.cond = PTHREAD_COND_INITIALIZER,
@@ -693,9 +690,6 @@ static int update_kernel_stream(struct consumer_data *consumer_data, int fd)
 					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,
 								channel, ksess,
@@ -730,6 +724,12 @@ static void update_ust_app(int app_sock)
 {
 	struct ltt_session *sess, *stmp;
 
+	/* Consumer is in an ERROR state. Stop any application update. */
+	if (uatomic_read(&ust_consumerd_state) == CONSUMER_ERROR) {
+		/* Stop the update process since the consumer is dead. */
+		return;
+	}
+
 	/* For all tracing session(s) */
 	cds_list_for_each_entry_safe(sess, stmp, &session_list_ptr->head, list) {
 		session_lock(sess);
@@ -1019,15 +1019,16 @@ restart:
 		/* Connect both socket, command and metadata. */
 		consumer_data->cmd_sock =
 			lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path);
-		consumer_data->metadata_sock.fd =
+		consumer_data->metadata_fd =
 			lttcomm_connect_unix_sock(consumer_data->cmd_unix_sock_path);
-		if (consumer_data->cmd_sock < 0 ||
-				consumer_data->metadata_sock.fd < 0) {
+		if (consumer_data->cmd_sock < 0
+				|| consumer_data->metadata_fd < 0) {
 			PERROR("consumer connect cmd socket");
 			/* On error, signal condition and quit. */
 			signal_consumer_condition(consumer_data, -1);
 			goto error;
 		}
+		consumer_data->metadata_sock.fd_ptr = &consumer_data->metadata_fd;
 		/* Create metadata socket lock. */
 		consumer_data->metadata_sock.lock = zmalloc(sizeof(pthread_mutex_t));
 		if (consumer_data->metadata_sock.lock == NULL) {
@@ -1040,7 +1041,7 @@ restart:
 		signal_consumer_condition(consumer_data, 1);
 		DBG("Consumer command socket ready (fd: %d", consumer_data->cmd_sock);
 		DBG("Consumer metadata socket ready (fd: %d)",
-				consumer_data->metadata_sock.fd);
+				consumer_data->metadata_fd);
 	} else {
 		ERR("consumer error when waiting for SOCK_READY : %s",
 				lttcomm_get_readable_code(-code));
@@ -1060,7 +1061,7 @@ restart:
 	}
 
 	/* Add metadata socket that is successfully connected. */
-	ret = lttng_poll_add(&events, consumer_data->metadata_sock.fd,
+	ret = lttng_poll_add(&events, consumer_data->metadata_fd,
 			LPOLLIN | LPOLLRDHUP);
 	if (ret < 0) {
 		goto error;
@@ -1119,7 +1120,7 @@ restart_poll:
 						lttcomm_get_readable_code(-code));
 
 				goto exit;
-			} else if (pollfd == consumer_data->metadata_sock.fd) {
+			} else if (pollfd == consumer_data->metadata_fd) {
 				/* UST metadata requests */
 				ret = ust_consumer_metadata_request(
 						&consumer_data->metadata_sock);
@@ -1138,6 +1139,13 @@ restart_poll:
 
 exit:
 error:
+	/*
+	 * We lock here because we are about to close the sockets and some other
+	 * thread might be using them so get exclusive access which will abort all
+	 * other consumer command by other threads.
+	 */
+	pthread_mutex_lock(&consumer_data->lock);
+
 	/* Immediately set the consumerd state to stopped */
 	if (consumer_data->type == LTTNG_CONSUMER_KERNEL) {
 		uatomic_set(&kernel_consumerd_state, CONSUMER_ERROR);
@@ -1163,15 +1171,12 @@ error:
 		}
 		consumer_data->cmd_sock = -1;
 	}
-	if (consumer_data->metadata_sock.fd >= 0) {
-		ret = close(consumer_data->metadata_sock.fd);
+	if (*consumer_data->metadata_sock.fd_ptr >= 0) {
+		ret = close(*consumer_data->metadata_sock.fd_ptr);
 		if (ret) {
 			PERROR("close");
 		}
 	}
-	/* Cleanup metadata socket mutex. */
-	pthread_mutex_destroy(consumer_data->metadata_sock.lock);
-	free(consumer_data->metadata_sock.lock);
 
 	if (sock >= 0) {
 		ret = close(sock);
@@ -1183,6 +1188,11 @@ error:
 	unlink(consumer_data->err_unix_sock_path);
 	unlink(consumer_data->cmd_unix_sock_path);
 	consumer_data->pid = 0;
+	pthread_mutex_unlock(&consumer_data->lock);
+
+	/* Cleanup metadata socket mutex. */
+	pthread_mutex_destroy(consumer_data->metadata_sock.lock);
+	free(consumer_data->metadata_sock.lock);
 
 	lttng_poll_clean(&events);
 error_poll:
@@ -1367,6 +1377,9 @@ error_testpoint:
  * Send a socket to a thread This is called from the dispatch UST registration
  * thread once all sockets are set for the application.
  *
+ * The sock value can be invalid, we don't really care, the thread will handle
+ * it and make the necessary cleanup if so.
+ *
  * On success, return 0 else a negative value being the errno message of the
  * write().
  */
@@ -1374,9 +1387,14 @@ 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);
+	/*
+	 * It's possible that the FD is set as invalid with -1 concurrently just
+	 * before calling this function being a shutdown state of the thread.
+	 */
+	if (fd < 0) {
+		ret = -EBADF;
+		goto error;
+	}
 
 	do {
 		ret = write(fd, &sock, sizeof(sock));
@@ -1639,7 +1657,12 @@ static void *thread_dispatch_ust_registration(void *data)
 				if (ret < 0) {
 					rcu_read_unlock();
 					session_unlock_list();
-					/* No notify thread, stop the UST tracing. */
+					/*
+					 * No notify thread, stop the UST tracing. However, this is
+					 * not an internal error of the this thread thus setting
+					 * the health error code to a normal exit.
+					 */
+					err = 0;
 					goto error;
 				}
 
@@ -1664,7 +1687,12 @@ static void *thread_dispatch_ust_registration(void *data)
 				if (ret < 0) {
 					rcu_read_unlock();
 					session_unlock_list();
-					/* No apps. thread, stop the UST tracing. */
+					/*
+					 * No apps. thread, stop the UST tracing. However, this is
+					 * not an internal error of the this thread thus setting
+					 * the health error code to a normal exit.
+					 */
+					err = 0;
 					goto error;
 				}
 
@@ -2440,6 +2468,7 @@ static int create_ust_session(struct ltt_session *session,
 	lus->gid = session->gid;
 	lus->output_traces = session->output_traces;
 	lus->snapshot_mode = session->snapshot_mode;
+	lus->live_timer_interval = session->live_timer;
 	session->ust_session = lus;
 
 	/* Copy session output to the newly created UST session */
@@ -2554,6 +2583,7 @@ 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_CREATE_SESSION_LIVE:
 	case LTTNG_DESTROY_SESSION:
 	case LTTNG_LIST_SESSIONS:
 	case LTTNG_LIST_DOMAINS:
@@ -2617,6 +2647,7 @@ 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_CREATE_SESSION_LIVE:
 	case LTTNG_CALIBRATE:
 	case LTTNG_LIST_SESSIONS:
 	case LTTNG_LIST_TRACEPOINTS:
@@ -2710,6 +2741,10 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
 		break;
 	case LTTNG_DOMAIN_UST:
 	{
+		if (!ust_app_supported()) {
+			ret = LTTNG_ERR_NO_UST;
+			goto error;
+		}
 		/* Consumer is in an ERROR state. Report back to client */
 		if (uatomic_read(&ust_consumerd_state) == CONSUMER_ERROR) {
 			ret = LTTNG_ERR_NO_USTCONSUMERD;
@@ -3057,7 +3092,7 @@ skip_domain:
 		}
 
 		ret = cmd_create_session_uri(cmd_ctx->lsm->session.name, uris, nb_uri,
-			&cmd_ctx->creds);
+			&cmd_ctx->creds, 0);
 
 		free(uris);
 
@@ -3347,6 +3382,45 @@ skip_domain:
 		free(uris);
 		break;
 	}
+	case LTTNG_CREATE_SESSION_LIVE:
+	{
+		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_uri(cmd_ctx->lsm->session.name, uris,
+				nb_uri, &cmd_ctx->creds, cmd_ctx->lsm->u.session_live.timer_interval);
+		free(uris);
+		break;
+	}
 	default:
 		ret = LTTNG_ERR_UND;
 		break;
@@ -4638,6 +4712,14 @@ int main(int argc, char **argv)
 
 	/* Initialize communication library */
 	lttcomm_init();
+	/* This is to get the TCP timeout value. */
+	lttcomm_inet_init();
+
+	/*
+	 * Initialize the health check subsystem. This call should set the
+	 * appropriate time values.
+	 */
+	health_init();
 
 	/* Create thread to manage the client socket */
 	ret = pthread_create(&ht_cleanup_thread, NULL,
@@ -4692,7 +4774,7 @@ int main(int argc, char **argv)
 			ust_thread_manage_notify, (void *) NULL);
 	if (ret != 0) {
 		PERROR("pthread_create apps");
-		goto exit_apps;
+		goto exit_apps_notify;
 	}
 
 	/* Don't start this thread if kernel tracing is not requested nor root */
@@ -4713,9 +4795,16 @@ int main(int argc, char **argv)
 	}
 
 exit_kernel:
+	ret = pthread_join(apps_notify_thread, &status);
+	if (ret != 0) {
+		PERROR("pthread_join apps notify");
+		goto error;	/* join error, exit without cleanup */
+	}
+
+exit_apps_notify:
 	ret = pthread_join(apps_thread, &status);
 	if (ret != 0) {
-		PERROR("pthread_join");
+		PERROR("pthread_join apps");
 		goto error;	/* join error, exit without cleanup */
 	}