+ /* Poll infinite value of time */
+ ret = poll(kernel_pollfd, nb_fd, -1);
+ if (ret < 0) {
+ perror("poll kernel thread");
+ goto error;
+ } else if (ret == 0) {
+ /* Should not happen since timeout is infinite */
+ continue;
+ }
+
+ /* Thread quit pipe has been closed. Killing thread. */
+ if (kernel_pollfd[nb_fd - 1].revents == POLLNVAL) {
+ goto error;
+ }
+
+ DBG("Kernel poll event triggered");
+
+ /*
+ * Check if the wake up pipe was triggered. If so, the kernel_pollfd
+ * must be updated.
+ */
+ switch (kernel_pollfd[nb_fd - 2].revents) {
+ case POLLIN:
+ ret = read(kernel_poll_pipe[0], &tmp, 1);
+ update_poll_flag = 1;
+ continue;
+ case POLLERR:
+ goto error;
+ default:
+ break;
+ }
+
+ for (i = 0; i < nb_fd; i++) {
+ switch (kernel_pollfd[i].revents) {
+ /*
+ * New CPU detected by the kernel. Adding kernel stream to kernel
+ * session and updating the kernel consumer
+ */
+ case POLLIN | POLLRDNORM:
+ ret = update_kernel_stream(kernel_pollfd[i].fd);
+ if (ret < 0) {
+ continue;
+ }
+ break;
+ }
+ }
+ }
+
+error:
+ DBG("Kernel thread dying");
+ if (kernel_pollfd) {
+ free(kernel_pollfd);
+ }
+
+ close(kernel_poll_pipe[0]);
+ close(kernel_poll_pipe[1]);
+ return NULL;
+}
+
+/*
+ * This thread manage the kconsumerd error sent back to the session daemon.
+ */
+static void *thread_manage_kconsumerd(void *data)
+{
+ int sock = 0, ret;
+ enum lttcomm_return_code code;
+ struct pollfd pollfd[2];
+
+ DBG("[thread] Manage kconsumerd started");
+
+ ret = lttcomm_listen_unix_sock(kconsumerd_err_sock);
+ if (ret < 0) {
+ goto error;
+ }
+
+ /* First fd is always the quit pipe */
+ pollfd[0].fd = thread_quit_pipe[0];
+
+ /* Apps socket */
+ pollfd[1].fd = kconsumerd_err_sock;
+ pollfd[1].events = POLLIN;
+
+ /* Inifinite blocking call, waiting for transmission */
+ ret = poll(pollfd, 2, -1);
+ if (ret < 0) {
+ perror("poll kconsumerd thread");
+ goto error;
+ }
+
+ /* Thread quit pipe has been closed. Killing thread. */
+ if (pollfd[0].revents == POLLNVAL) {
+ goto error;
+ } else if (pollfd[1].revents == POLLERR) {
+ ERR("Kconsumerd err socket poll error");
+ goto error;
+ }
+
+ sock = lttcomm_accept_unix_sock(kconsumerd_err_sock);
+ if (sock < 0) {
+ goto error;
+ }
+
+ /* Getting status code from kconsumerd */
+ ret = lttcomm_recv_unix_sock(sock, &code, sizeof(enum lttcomm_return_code));
+ if (ret <= 0) {
+ goto error;
+ }
+
+ if (code == KCONSUMERD_COMMAND_SOCK_READY) {
+ kconsumerd_cmd_sock = lttcomm_connect_unix_sock(kconsumerd_cmd_unix_sock_path);
+ if (kconsumerd_cmd_sock < 0) {
+ sem_post(&kconsumerd_sem);
+ perror("kconsumerd connect");
+ goto error;
+ }
+ /* Signal condition to tell that the kconsumerd is ready */
+ sem_post(&kconsumerd_sem);
+ DBG("Kconsumerd command socket ready");
+ } else {
+ DBG("Kconsumerd error when waiting for SOCK_READY : %s",
+ lttcomm_get_readable_code(-code));
+ goto error;
+ }
+
+ /* Kconsumerd err socket */
+ pollfd[1].fd = sock;
+ pollfd[1].events = POLLIN;
+
+ /* Inifinite blocking call, waiting for transmission */
+ ret = poll(pollfd, 2, -1);
+ if (ret < 0) {
+ perror("poll kconsumerd thread");
+ goto error;
+ }
+
+ /* Thread quit pipe has been closed. Killing thread. */
+ if (pollfd[0].revents == POLLNVAL) {
+ goto error;
+ } else if (pollfd[1].revents == POLLERR) {
+ ERR("Kconsumerd err socket second poll error");
+ goto error;
+ }
+
+ /* Wait for any kconsumerd error */
+ ret = lttcomm_recv_unix_sock(sock, &code, sizeof(enum lttcomm_return_code));
+ if (ret <= 0) {
+ ERR("Kconsumerd closed the command socket");
+ goto error;
+ }
+
+ ERR("Kconsumerd return code : %s", lttcomm_get_readable_code(-code));
+
+error:
+ DBG("Kconsumerd thread dying");
+ if (kconsumerd_err_sock) {
+ close(kconsumerd_err_sock);
+ }
+ if (kconsumerd_cmd_sock) {
+ close(kconsumerd_cmd_sock);
+ }
+ if (sock) {
+ close(sock);
+ }
+
+ unlink(kconsumerd_err_unix_sock_path);
+ unlink(kconsumerd_cmd_unix_sock_path);
+
+ kconsumerd_pid = 0;
+ return NULL;
+}
+
+/*
+ * This thread manage the application socket communication
+ */
+static void *thread_manage_apps(void *data)
+{
+ int sock = 0, ret;
+ struct pollfd pollfd[2];
+
+ /* TODO: Something more elegant is needed but fine for now */
+ /* FIXME: change all types to either uint8_t, uint32_t, uint64_t
+ * for 32-bit vs 64-bit compat processes. */
+ /* replicate in ust with version number */
+ struct {
+ int reg; /* 1:register, 0:unregister */
+ pid_t pid;
+ uid_t uid;
+ } reg_msg;
+
+ DBG("[thread] Manage apps started");
+
+ ret = lttcomm_listen_unix_sock(apps_sock);
+ if (ret < 0) {
+ goto error;
+ }
+
+ /* First fd is always the quit pipe */
+ pollfd[0].fd = thread_quit_pipe[0];
+
+ /* Apps socket */
+ pollfd[1].fd = apps_sock;
+ pollfd[1].events = POLLIN;
+
+ /* Notify all applications to register */
+ notify_apps(default_global_apps_pipe);
+
+ while (1) {
+ DBG("Accepting application registration");
+
+ /* Inifinite blocking call, waiting for transmission */
+ ret = poll(pollfd, 2, -1);
+ if (ret < 0) {
+ perror("poll apps thread");
+ goto error;
+ }
+
+ /* Thread quit pipe has been closed. Killing thread. */
+ if (pollfd[0].revents == POLLNVAL) {
+ goto error;
+ } else if (pollfd[1].revents == POLLERR) {
+ ERR("Apps socket poll error");
+ goto error;
+ }
+
+ sock = lttcomm_accept_unix_sock(apps_sock);
+ if (sock < 0) {
+ goto error;
+ }
+
+ /*
+ * Using message-based transmissions to ensure we don't
+ * have to deal with partially received messages.
+ */
+ ret = lttcomm_recv_unix_sock(sock, ®_msg, sizeof(reg_msg));
+ if (ret < 0) {
+ perror("recv");
+ continue;
+ }
+
+ /* Add application to the global traceable list */
+ if (reg_msg.reg == 1) {
+ /* Registering */
+ /*
+ * TODO: socket should be either passed to a
+ * listener thread (for more messages) or
+ * closed. It currently leaks.
+ */
+ ret = register_traceable_app(reg_msg.pid, reg_msg.uid);
+ if (ret < 0) {
+ /* register_traceable_app only return an error with
+ * ENOMEM. At this point, we better stop everything.
+ */
+ goto error;
+ }
+ } else {
+ /* Unregistering */
+ unregister_traceable_app(reg_msg.pid);
+ }
+ }
+
+error:
+ DBG("Apps thread dying");
+ if (apps_sock) {
+ close(apps_sock);
+ }
+ if (sock) {
+ close(sock);
+ }
+
+ unlink(apps_unix_sock_path);
+ return NULL;
+}
+
+/*
+ * Start the thread_manage_kconsumerd. This must be done after a kconsumerd
+ * exec or it will fails.
+ */
+static int spawn_kconsumerd_thread(void)
+{
+ int ret;
+
+ /* Setup semaphore */
+ sem_init(&kconsumerd_sem, 0, 0);
+
+ ret = pthread_create(&kconsumerd_thread, NULL, thread_manage_kconsumerd, (void *) NULL);
+ if (ret != 0) {
+ perror("pthread_create kconsumerd");
+ goto error;
+ }
+
+ /* Wait for the kconsumerd thread to be ready */
+ sem_wait(&kconsumerd_sem);
+
+ if (kconsumerd_pid == 0) {
+ ERR("Kconsumerd did not start");
+ goto error;
+ }
+
+ return 0;
+
+error:
+ ret = LTTCOMM_KERN_CONSUMER_FAIL;
+ return ret;
+}
+
+/*
+ * Join kernel consumer thread
+ */
+static int join_kconsumerd_thread(void)
+{
+ void *status;
+ int ret;
+
+ if (kconsumerd_pid != 0) {
+ ret = kill(kconsumerd_pid, SIGTERM);
+ if (ret) {
+ ERR("Error killing kconsumerd");
+ return ret;
+ }
+ return pthread_join(kconsumerd_thread, &status);
+ } else {
+ return 0;
+ }
+}
+
+/*
+ * Fork and exec a kernel consumer daemon (kconsumerd).
+ *
+ * Return pid if successful else -1.
+ */
+static pid_t spawn_kconsumerd(void)
+{
+ int ret;
+ pid_t pid;
+ const char *verbosity;
+
+ DBG("Spawning kconsumerd");
+
+ pid = fork();
+ if (pid == 0) {
+ /*
+ * Exec kconsumerd.
+ */
+ if (opt_verbose > 1 || opt_verbose_kconsumerd) {
+ verbosity = "--verbose";
+ } else {
+ verbosity = "--quiet";
+ }
+ execl(INSTALL_BIN_PATH "/ltt-kconsumerd", "ltt-kconsumerd", verbosity, NULL);
+ if (errno != 0) {
+ perror("kernel start consumer exec");
+ }
+ exit(EXIT_FAILURE);
+ } else if (pid > 0) {
+ ret = pid;
+ goto error;
+ } else {
+ perror("kernel start consumer fork");
+ ret = -errno;
+ goto error;
+ }
+
+error:
+ return ret;
+}
+
+/*
+ * Spawn the kconsumerd daemon and session daemon thread.
+ */
+static int start_kconsumerd(void)
+{
+ int ret;
+
+ pthread_mutex_lock(&kconsumerd_pid_mutex);
+ if (kconsumerd_pid != 0) {
+ pthread_mutex_unlock(&kconsumerd_pid_mutex);
+ goto end;
+ }
+
+ ret = spawn_kconsumerd();
+ if (ret < 0) {
+ ERR("Spawning kconsumerd failed");
+ ret = LTTCOMM_KERN_CONSUMER_FAIL;
+ pthread_mutex_unlock(&kconsumerd_pid_mutex);
+ goto error;
+ }
+
+ /* Setting up the global kconsumerd_pid */
+ kconsumerd_pid = ret;
+ pthread_mutex_unlock(&kconsumerd_pid_mutex);
+
+ DBG("Kconsumerd pid %d", ret);
+
+ DBG("Spawning kconsumerd thread");
+ ret = spawn_kconsumerd_thread();
+ if (ret < 0) {
+ ERR("Fatal error spawning kconsumerd thread");
+ goto error;
+ }
+
+end: