+ /* 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 register 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("Register apps socket poll error");
+ goto error;
+ }
+
+ sock = lttcomm_accept_unix_sock(apps_sock);
+ if (sock < 0) {
+ goto error;
+ }
+
+ /* Create UST registration command for enqueuing */
+ ust_cmd = malloc(sizeof(struct ust_command));
+ if (ust_cmd == NULL) {
+ perror("ust command malloc");
+ goto error;
+ }
+
+ /*
+ * Using message-based transmissions to ensure we don't have to deal
+ * with partially received messages.
+ */
+ ret = lttcomm_recv_unix_sock(sock, &ust_cmd->reg_msg,
+ sizeof(struct ust_register_msg));
+ if (ret < 0 || ret < sizeof(struct ust_register_msg)) {
+ if (ret < 0) {
+ perror("lttcomm_recv_unix_sock register apps");
+ } else {
+ ERR("Wrong size received on apps register");
+ }
+ free(ust_cmd);
+ close(sock);
+ continue;
+ }
+
+ ust_cmd->sock = sock;
+
+ DBG("UST registration received with pid:%d ppid:%d uid:%d"
+ " gid:%d sock:%d name:%s (version %d.%d)",
+ ust_cmd->reg_msg.pid, ust_cmd->reg_msg.ppid,
+ ust_cmd->reg_msg.uid, ust_cmd->reg_msg.gid,
+ ust_cmd->sock, ust_cmd->reg_msg.name,
+ ust_cmd->reg_msg.major, ust_cmd->reg_msg.minor);
+ /*
+ * Lock free enqueue the registration request.
+ * The red pill has been taken! This apps will be part of the *system*
+ */
+ cds_wfq_enqueue(&ust_cmd_queue.queue, &ust_cmd->node);
+
+ /*
+ * Wake the registration queue futex.
+ * Implicit memory barrier with the exchange in cds_wfq_enqueue.
+ */
+ futex_nto1_wake(&ust_cmd_queue.futex);
+ }
+
+error:
+ DBG("Register 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;