X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fmain.c;h=476b1b35fcbd65fcf6daeeb829309ad7613900e6;hb=25c629438a161e3ae35fa50baf63339b031bd749;hp=2582e886d1e1a5f5352a6f91b8bd82b2ea914c7b;hpb=77cdfc81210ea4cfca740335af03be732e652e6d;p=lttng-tools.git

diff --git a/src/bin/lttng-sessiond/main.c b/src/bin/lttng-sessiond/main.c
index 2582e886d..476b1b35f 100644
--- a/src/bin/lttng-sessiond/main.c
+++ b/src/bin/lttng-sessiond/main.c
@@ -582,6 +582,34 @@ static int generate_lock_file_path(char *path, size_t len)
 	return ret;
 }
 
+/*
+ * Wait on consumer process termination.
+ *
+ * Need to be called with the consumer data lock held or from a context
+ * ensuring no concurrent access to data (e.g: cleanup).
+ */
+static void wait_consumer(struct consumer_data *consumer_data)
+{
+	pid_t ret;
+	int status;
+
+	if (consumer_data->pid <= 0) {
+		return;
+	}
+
+	DBG("Waiting for complete teardown of consumerd (PID: %d)",
+			consumer_data->pid);
+	ret = waitpid(consumer_data->pid, &status, 0);
+	if (ret == -1) {
+		PERROR("consumerd waitpid pid: %d", consumer_data->pid)
+	}
+	if (!WIFEXITED(status)) {
+		ERR("consumerd termination with error: %d",
+				WEXITSTATUS(ret));
+	}
+	consumer_data->pid = 0;
+}
+
 /*
  * Cleanup the session daemon's data structures.
  */
@@ -676,6 +704,10 @@ static void sessiond_cleanup(void)
 		}
 	}
 
+	wait_consumer(&kconsumer_data);
+	wait_consumer(&ustconsumer64_data);
+	wait_consumer(&ustconsumer32_data);
+
 	DBG("Cleaning up all agent apps");
 	agent_app_ht_clean();
 
@@ -765,6 +797,8 @@ static void sessiond_cleanup_options(void)
 	free(kmod_probes_list);
 	free(kmod_extra_probes_list);
 
+	run_as_destroy_worker();
+
 	/* <fun> */
 	DBG("%c[%d;%dm*** assert failed :-) *** ==> %c[%dm%c[%d;%dm"
 			"Matthew, BEET driven development works!%c[%dm",
@@ -1129,31 +1163,33 @@ static void *thread_manage_kernel(void *data)
 			}
 
 			/* Check for data on kernel pipe */
-			if (pollfd == kernel_poll_pipe[0] && (revents & LPOLLIN)) {
-				(void) lttng_read(kernel_poll_pipe[0],
-					&tmp, 1);
-				/*
-				 * Ret value is useless here, if this pipe gets any actions an
-				 * update is required anyway.
-				 */
-				update_poll_flag = 1;
-				continue;
-			} else {
-				/*
-				 * New CPU detected by the kernel. Adding kernel stream to
-				 * kernel session and updating the kernel consumer
-				 */
-				if (revents & LPOLLIN) {
+			if (revents & LPOLLIN) {
+				if (pollfd == kernel_poll_pipe[0]) {
+					(void) lttng_read(kernel_poll_pipe[0],
+						&tmp, 1);
+					/*
+					 * Ret value is useless here, if this pipe gets any actions an
+					 * update is required anyway.
+					 */
+					update_poll_flag = 1;
+					continue;
+				} else {
+					/*
+					 * New CPU detected by the kernel. Adding kernel stream to
+					 * kernel session and updating the kernel consumer
+					 */
 					ret = update_kernel_stream(&kconsumer_data, pollfd);
 					if (ret < 0) {
 						continue;
 					}
 					break;
-					/*
-					 * TODO: We might want to handle the LPOLLERR | LPOLLHUP
-					 * and unregister kernel stream at this point.
-					 */
 				}
+			} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+				update_poll_flag = 1;
+				continue;
+			} else {
+				ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+				goto error;
 			}
 		}
 	}
@@ -1212,6 +1248,9 @@ static void *thread_manage_consumer(void *data)
 
 	DBG("[thread] Manage consumer started");
 
+	rcu_register_thread();
+	rcu_thread_online();
+
 	health_register(health_sessiond, HEALTH_SESSIOND_TYPE_CONSUMER);
 
 	health_code_update();
@@ -1280,9 +1319,14 @@ restart:
 
 		/* Event on the registration socket */
 		if (pollfd == consumer_data->err_sock) {
-			if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+			if (revents & LPOLLIN) {
+				continue;
+			} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 				ERR("consumer err socket poll error");
 				goto error;
+			} else {
+				ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+				goto error;
 			}
 		}
 	}
@@ -1412,7 +1456,8 @@ restart_poll:
 
 			if (pollfd == sock) {
 				/* Event on the consumerd socket */
-				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)
+						&& !(revents & LPOLLIN)) {
 					ERR("consumer err socket second poll error");
 					goto error;
 				}
@@ -1430,6 +1475,11 @@ restart_poll:
 
 				goto exit;
 			} else if (pollfd == consumer_data->metadata_fd) {
+				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)
+						&& !(revents & LPOLLIN)) {
+					ERR("consumer err metadata socket second poll error");
+					goto error;
+				}
 				/* UST metadata requests */
 				ret = ust_consumer_metadata_request(
 						&consumer_data->metadata_sock);
@@ -1493,7 +1543,6 @@ 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. */
@@ -1510,6 +1559,9 @@ error_poll:
 	health_unregister(health_sessiond);
 	DBG("consumer thread cleanup completed");
 
+	rcu_thread_offline();
+	rcu_unregister_thread();
+
 	return NULL;
 }
 
@@ -1595,10 +1647,7 @@ static void *thread_manage_apps(void *data)
 
 			/* Inspect the apps cmd pipe */
 			if (pollfd == apps_cmd_pipe[0]) {
-				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
-					ERR("Apps command pipe error");
-					goto error;
-				} else if (revents & LPOLLIN) {
+				if (revents & LPOLLIN) {
 					int sock;
 
 					/* Empty pipe */
@@ -1611,9 +1660,8 @@ static void *thread_manage_apps(void *data)
 					health_code_update();
 
 					/*
-					 * We only monitor the error events of the socket. This
-					 * thread does not handle any incoming data from UST
-					 * (POLLIN).
+					 * Since this is a command socket (write then read),
+					 * we only monitor the error events of the socket.
 					 */
 					ret = lttng_poll_add(&events, sock,
 							LPOLLERR | LPOLLHUP | LPOLLRDHUP);
@@ -1622,6 +1670,12 @@ static void *thread_manage_apps(void *data)
 					}
 
 					DBG("Apps with sock %d added to poll set", sock);
+				} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+					ERR("Apps command pipe error");
+					goto error;
+				} else {
+					ERR("Unknown poll events %u for sock %d", revents, pollfd);
+					goto error;
 				}
 			} else {
 				/*
@@ -1637,6 +1691,9 @@ static void *thread_manage_apps(void *data)
 
 					/* Socket closed on remote end. */
 					ust_app_unregister(pollfd);
+				} else {
+					ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+					goto error;
 				}
 			}
 
@@ -1778,6 +1835,9 @@ static void sanitize_wait_queue(struct ust_reg_wait_queue *wait_queue)
 				ust_app_destroy(wait_node->app);
 				free(wait_node);
 				break;
+			} else {
+				ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+				goto error;
 			}
 		}
 	}
@@ -1979,7 +2039,7 @@ static void *thread_dispatch_ust_registration(void *data)
 				 * Don't care about return value. Let the manage apps threads
 				 * handle app unregistration upon socket close.
 				 */
-				(void) ust_app_register_done(app->sock);
+				(void) ust_app_register_done(app);
 
 				/*
 				 * Even if the application socket has been closed, send the app
@@ -2021,6 +2081,22 @@ error:
 		free(wait_node);
 	}
 
+	/* Empty command queue. */
+	for (;;) {
+		/* Dequeue command for registration */
+		node = cds_wfcq_dequeue_blocking(&ust_cmd_queue.head, &ust_cmd_queue.tail);
+		if (node == NULL) {
+			break;
+		}
+		ust_cmd = caa_container_of(node, struct ust_command, node);
+		ret = close(ust_cmd->sock);
+		if (ret < 0) {
+			PERROR("close ust sock exit dispatch %d", ust_cmd->sock);
+		}
+		lttng_fd_put(LTTNG_FD_APPS, 1);
+		free(ust_cmd);
+	}
+
 error_testpoint:
 	DBG("Dispatch thread dying");
 	if (err) {
@@ -2122,10 +2198,7 @@ static void *thread_registration_apps(void *data)
 
 			/* Event on the registration socket */
 			if (pollfd == apps_sock) {
-				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
-					ERR("Register apps socket poll error");
-					goto error;
-				} else if (revents & LPOLLIN) {
+				if (revents & LPOLLIN) {
 					sock = lttcomm_accept_unix_sock(apps_sock);
 					if (sock < 0) {
 						goto error;
@@ -2138,10 +2211,12 @@ static void *thread_registration_apps(void *data)
 					 * lttcomm_setsockopt_snd_timeout expect msec as
 					 * parameter.
 					 */
-					(void) lttcomm_setsockopt_rcv_timeout(sock,
-							app_socket_timeout * 1000);
-					(void) lttcomm_setsockopt_snd_timeout(sock,
-							app_socket_timeout * 1000);
+					if (app_socket_timeout >= 0) {
+						(void) lttcomm_setsockopt_rcv_timeout(sock,
+								app_socket_timeout * 1000);
+						(void) lttcomm_setsockopt_snd_timeout(sock,
+								app_socket_timeout * 1000);
+					}
 
 					/*
 					 * Set the CLOEXEC flag. Return code is useless because
@@ -2153,6 +2228,10 @@ static void *thread_registration_apps(void *data)
 					ust_cmd = zmalloc(sizeof(struct ust_command));
 					if (ust_cmd == NULL) {
 						PERROR("ust command zmalloc");
+						ret = close(sock);
+						if (ret) {
+							PERROR("close");
+						}
 						goto error;
 					}
 
@@ -2208,6 +2287,12 @@ static void *thread_registration_apps(void *data)
 					 * barrier with the exchange in cds_wfcq_enqueue.
 					 */
 					futex_nto1_wake(&ust_cmd_queue.futex);
+				} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+					ERR("Register apps socket poll error");
+					goto error;
+				} else {
+					ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+					goto error;
 				}
 			}
 		}
@@ -2709,7 +2794,7 @@ static int copy_session_consumer(int domain, struct ltt_session *session)
 		 * domain.
 		 */
 		if (session->kernel_session->consumer) {
-			consumer_destroy_output(session->kernel_session->consumer);
+			consumer_output_put(session->kernel_session->consumer);
 		}
 		session->kernel_session->consumer =
 			consumer_copy_output(session->consumer);
@@ -2723,7 +2808,7 @@ static int copy_session_consumer(int domain, struct ltt_session *session)
 	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);
+			consumer_output_put(session->ust_session->consumer);
 		}
 		session->ust_session->consumer =
 			consumer_copy_output(session->consumer);
@@ -2843,7 +2928,7 @@ static int create_kernel_session(struct ltt_session *session)
 				session->kernel_session->consumer->dst.trace_path,
 				S_IRWXU | S_IRWXG, session->uid, session->gid);
 		if (ret < 0) {
-			if (ret != -EEXIST) {
+			if (errno != EEXIST) {
 				ERR("Trace directory creation error");
 				goto error;
 			}
@@ -2905,6 +2990,8 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
 
 	DBG("Processing client command %d", cmd_ctx->lsm->cmd_type);
 
+	assert(!rcu_read_ongoing());
+
 	*sock_error = 0;
 
 	switch (cmd_ctx->lsm->cmd_type) {
@@ -2964,6 +3051,7 @@ static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
 	case LTTNG_LIST_EVENTS:
 	case LTTNG_LIST_SYSCALLS:
 	case LTTNG_LIST_TRACKER_PIDS:
+	case LTTNG_DATA_PENDING:
 		break;
 	default:
 		/* Setup lttng message with no payload */
@@ -3200,6 +3288,8 @@ skip_domain:
 	if (cmd_ctx->lsm->cmd_type == LTTNG_START_TRACE ||
 			cmd_ctx->lsm->cmd_type == LTTNG_STOP_TRACE) {
 		switch (cmd_ctx->lsm->domain.type) {
+		case LTTNG_DOMAIN_NONE:
+			break;
 		case LTTNG_DOMAIN_JUL:
 		case LTTNG_DOMAIN_LOG4J:
 		case LTTNG_DOMAIN_PYTHON:
@@ -3215,6 +3305,9 @@ skip_domain:
 				goto error;
 			}
 			break;
+		default:
+			ret = LTTNG_ERR_UNKNOWN_DOMAIN;
+			goto error;
 		}
 	}
 
@@ -3263,8 +3356,34 @@ skip_domain:
 	}
 	case LTTNG_DISABLE_EVENT:
 	{
+
+		/*
+		 * FIXME: handle filter; for now we just receive the filter's
+		 * bytecode along with the filter expression which are sent by
+		 * liblttng-ctl and discard them.
+		 *
+		 * This fixes an issue where the client may block while sending
+		 * the filter payload and encounter an error because the session
+		 * daemon closes the socket without ever handling this data.
+		 */
+		size_t count = cmd_ctx->lsm->u.disable.expression_len +
+			cmd_ctx->lsm->u.disable.bytecode_len;
+
+		if (count) {
+			char data[LTTNG_FILTER_MAX_LEN];
+
+			DBG("Discarding disable event command payload of size %zu", count);
+			while (count) {
+				ret = lttcomm_recv_unix_sock(sock, data,
+				        count > sizeof(data) ? sizeof(data) : count);
+				if (ret < 0) {
+					goto error;
+				}
+
+				count -= (size_t) ret;
+			}
+		}
 		/* FIXME: passing packed structure to non-packed pointer */
-		/* TODO: handle filter */
 		ret = cmd_disable_event(cmd_ctx->session, cmd_ctx->lsm->domain.type,
 				cmd_ctx->lsm->u.disable.channel_name,
 				&cmd_ctx->lsm->u.disable.event);
@@ -3938,6 +4057,7 @@ setup_error:
 		session_unlock_list();
 	}
 init_setup_error:
+	assert(!rcu_read_ongoing());
 	return ret;
 }
 
@@ -4053,9 +4173,14 @@ restart:
 
 			/* Event on the registration socket */
 			if (pollfd == sock) {
-				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+				if (revents & LPOLLIN) {
+					continue;
+				} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 					ERR("Health socket poll error");
 					goto error;
+				} else {
+					ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+					goto error;
 				}
 			}
 		}
@@ -4126,7 +4251,7 @@ error:
 	}
 
 	lttng_poll_clean(&events);
-
+	stop_threads();
 	rcu_unregister_thread();
 	return NULL;
 }
@@ -4230,9 +4355,14 @@ static void *thread_manage_clients(void *data)
 
 			/* Event on the registration socket */
 			if (pollfd == client_sock) {
-				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+				if (revents & LPOLLIN) {
+					continue;
+				} else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 					ERR("Client socket poll error");
 					goto error;
+				} else {
+					ERR("Unexpected poll events %u for sock %d", revents, pollfd);
+					goto error;
 				}
 			}
 		}
@@ -4458,18 +4588,11 @@ static int set_option(int opt, const char *arg, const char *optname)
 {
 	int ret = 0;
 
-	if (arg && arg[0] == '\0') {
-		/*
-		 * This only happens if the value is read from daemon config
-		 * file. This means the option requires an argument and the
-		 * configuration file contains a line such as:
-		 * my_option =
-		 */
-		ret = -EINVAL;
-		goto end;
-	}
-
 	if (string_match(optname, "client-sock") || opt == 'c') {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"-c, --client-sock");
@@ -4477,6 +4600,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			snprintf(client_unix_sock_path, PATH_MAX, "%s", arg);
 		}
 	} else if (string_match(optname, "apps-sock") || opt == 'a') {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"-a, --apps-sock");
@@ -4488,6 +4615,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 	} else if (string_match(optname, "background") || opt == 'b') {
 		opt_background = 1;
 	} else if (string_match(optname, "group") || opt == 'g') {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"-g, --group");
@@ -4516,6 +4647,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 	} else if (string_match(optname, "sig-parent") || opt == 'S') {
 		opt_sig_parent = 1;
 	} else if (string_match(optname, "kconsumerd-err-sock")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--kconsumerd-err-sock");
@@ -4523,6 +4658,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			snprintf(kconsumer_data.err_unix_sock_path, PATH_MAX, "%s", arg);
 		}
 	} else if (string_match(optname, "kconsumerd-cmd-sock")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--kconsumerd-cmd-sock");
@@ -4530,6 +4669,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			snprintf(kconsumer_data.cmd_unix_sock_path, PATH_MAX, "%s", arg);
 		}
 	} else if (string_match(optname, "ustconsumerd64-err-sock")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--ustconsumerd64-err-sock");
@@ -4537,6 +4680,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			snprintf(ustconsumer64_data.err_unix_sock_path, PATH_MAX, "%s", arg);
 		}
 	} else if (string_match(optname, "ustconsumerd64-cmd-sock")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--ustconsumerd64-cmd-sock");
@@ -4544,6 +4691,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			snprintf(ustconsumer64_data.cmd_unix_sock_path, PATH_MAX, "%s", arg);
 		}
 	} else if (string_match(optname, "ustconsumerd32-err-sock")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--ustconsumerd32-err-sock");
@@ -4551,6 +4702,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			snprintf(ustconsumer32_data.err_unix_sock_path, PATH_MAX, "%s", arg);
 		}
 	} else if (string_match(optname, "ustconsumerd32-cmd-sock")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--ustconsumerd32-cmd-sock");
@@ -4580,6 +4735,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			opt_verbose_consumer += 1;
 		}
 	} else if (string_match(optname, "consumerd32-path")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--consumerd32-path");
@@ -4595,6 +4754,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			consumerd32_bin_override = 1;
 		}
 	} else if (string_match(optname, "consumerd32-libdir")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--consumerd32-libdir");
@@ -4610,6 +4773,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			consumerd32_libdir_override = 1;
 		}
 	} else if (string_match(optname, "consumerd64-path")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--consumerd64-path");
@@ -4625,6 +4792,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			consumerd64_bin_override = 1;
 		}
 	} else if (string_match(optname, "consumerd64-libdir")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--consumerd64-libdir");
@@ -4640,6 +4811,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			consumerd64_libdir_override = 1;
 		}
 	} else if (string_match(optname, "pidfile") || opt == 'p') {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"-p, --pidfile");
@@ -4652,6 +4827,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			}
 		}
 	} else if (string_match(optname, "agent-tcp-port")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--agent-tcp-port");
@@ -4676,6 +4855,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			DBG3("Agent TCP port set to non default: %u", agent_tcp_port);
 		}
 	} else if (string_match(optname, "load") || opt == 'l') {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"-l, --load");
@@ -4688,6 +4871,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			}
 		}
 	} else if (string_match(optname, "kmod-probes")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--kmod-probes");
@@ -4700,6 +4887,10 @@ static int set_option(int opt, const char *arg, const char *optname)
 			}
 		}
 	} else if (string_match(optname, "extra-kmod-probes")) {
+		if (!arg || *arg == '\0') {
+			ret = -EINVAL;
+			goto end;
+		}
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 				"--extra-kmod-probes");
@@ -5134,9 +5325,6 @@ error:
 static void sighandler(int sig)
 {
 	switch (sig) {
-	case SIGPIPE:
-		DBG("SIGPIPE caught");
-		return;
 	case SIGINT:
 		DBG("SIGINT caught");
 		stop_threads();
@@ -5168,9 +5356,10 @@ static int set_signal_handler(void)
 		return ret;
 	}
 
-	sa.sa_handler = sighandler;
 	sa.sa_mask = sigset;
 	sa.sa_flags = 0;
+
+	sa.sa_handler = sighandler;
 	if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
 		PERROR("sigaction");
 		return ret;
@@ -5181,12 +5370,13 @@ static int set_signal_handler(void)
 		return ret;
 	}
 
-	if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
+	if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
 		PERROR("sigaction");
 		return ret;
 	}
 
-	if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
+	sa.sa_handler = SIG_IGN;
+	if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
 		PERROR("sigaction");
 		return ret;
 	}
@@ -5226,7 +5416,10 @@ static int write_pidfile(void)
 	assert(rundir);
 
 	if (opt_pidfile) {
-		strncpy(pidfile_path, opt_pidfile, sizeof(pidfile_path));
+		if (lttng_strncpy(pidfile_path, opt_pidfile, sizeof(pidfile_path))) {
+			ret = -1;
+			goto error;
+		}
 	} else {
 		/* Build pidfile path from rundir and opt_pidfile. */
 		ret = snprintf(pidfile_path, sizeof(pidfile_path), "%s/"
@@ -5354,6 +5547,10 @@ int main(int argc, char **argv)
 		}
 	}
 
+	if (run_as_create_worker(argv[0]) < 0) {
+		goto exit_create_run_as_worker_cleanup;
+	}
+
 	/*
 	 * Starting from here, we can create threads. This needs to be after
 	 * lttng_daemonize due to RCU.
@@ -5907,6 +6104,10 @@ exit_apps:
 	}
 exit_reg_apps:
 
+	/*
+	 * Join dispatch thread after joining reg_apps_thread to ensure
+	 * we don't leak applications in the queue.
+	 */
 	ret = pthread_join(dispatch_thread, &status);
 	if (ret) {
 		errno = ret;
@@ -5941,6 +6142,13 @@ exit_init_data:
 	rcu_thread_offline();
 	rcu_unregister_thread();
 
+	/*
+	 * Ensure all prior call_rcu are done. call_rcu callbacks may push
+	 * hash tables to the ht_cleanup thread. Therefore, we ensure that
+	 * the queue is empty before shutting down the clean-up thread.
+	 */
+	rcu_barrier();
+
 	ret = notify_thread_pipe(ht_cleanup_quit_pipe[1]);
 	if (ret < 0) {
 		ERR("write error on ht_cleanup quit pipe");
@@ -5967,11 +6175,13 @@ exit_ht_cleanup_quit_pipe:
 
 	health_app_destroy(health_sessiond);
 exit_health_sessiond_cleanup:
+exit_create_run_as_worker_cleanup:
 
 exit_options:
 	sessiond_cleanup_options();
 
 exit_set_signal_handler:
+
 	if (!retval) {
 		exit(EXIT_SUCCESS);
 	} else {