+ rcu_register_thread();
+
+ DBG("Thread metadata poll started");
+
+ /* Size is set to 1 for the consumer_metadata pipe */
+ ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
+ if (ret < 0) {
+ ERR("Poll set creation failed");
+ goto end;
+ }
+
+ ret = lttng_poll_add(&events, ctx->consumer_metadata_pipe[0], LPOLLIN);
+ if (ret < 0) {
+ goto end;
+ }
+
+ /* Main loop */
+ DBG("Metadata main loop started");
+
+ while (1) {
+ lttng_poll_reset(&events);
+
+ nb_fd = LTTNG_POLL_GETNB(&events);
+
+ /* Only the metadata pipe is set */
+ if (nb_fd == 0 && consumer_quit == 1) {
+ goto end;
+ }
+
+restart:
+ DBG("Metadata poll wait with %d fd(s)", nb_fd);
+ ret = lttng_poll_wait(&events, -1);
+ DBG("Metadata event catched in thread");
+ if (ret < 0) {
+ if (errno == EINTR) {
+ ERR("Poll EINTR catched");
+ goto restart;
+ }
+ goto error;
+ }
+
+ /* From here, the event is a metadata wait fd */
+ for (i = 0; i < nb_fd; i++) {
+ revents = LTTNG_POLL_GETEV(&events, i);
+ pollfd = LTTNG_POLL_GETFD(&events, i);
+
+ /* Just don't waste time if no returned events for the fd */
+ if (!revents) {
+ continue;
+ }
+
+ if (pollfd == ctx->consumer_metadata_pipe[0]) {
+ if (revents & (LPOLLERR | LPOLLHUP )) {
+ DBG("Metadata thread pipe hung up");
+ /*
+ * Remove the pipe from the poll set and continue the loop
+ * since their might be data to consume.
+ */
+ lttng_poll_del(&events, ctx->consumer_metadata_pipe[0]);
+ ret = close(ctx->consumer_metadata_pipe[0]);
+ if (ret < 0) {
+ PERROR("close metadata pipe");
+ }
+ continue;
+ } else if (revents & LPOLLIN) {
+ do {
+ /* Get the stream pointer received */
+ ret = read(pollfd, &stream, sizeof(stream));
+ } while (ret < 0 && errno == EINTR);
+ if (ret < 0 ||
+ ret < sizeof(struct lttng_consumer_stream *)) {
+ PERROR("read metadata stream");
+ /*
+ * Let's continue here and hope we can still work
+ * without stopping the consumer. XXX: Should we?
+ */
+ continue;
+ }
+
+ /* A NULL stream means that the state has changed. */
+ if (stream == NULL) {
+ /* Check for deleted streams. */
+ validate_endpoint_status_metadata_stream(&events);
+ continue;
+ }
+
+ DBG("Adding metadata stream %d to poll set",
+ stream->wait_fd);
+
+ ret = consumer_add_metadata_stream(stream, metadata_ht);
+ if (ret) {
+ ERR("Unable to add metadata stream");
+ /* Stream was not setup properly. Continuing. */
+ consumer_del_metadata_stream(stream, NULL);
+ continue;
+ }
+
+ /* Add metadata stream to the global poll events list */
+ lttng_poll_add(&events, stream->wait_fd,
+ LPOLLIN | LPOLLPRI);
+ }
+
+ /* Handle other stream */
+ continue;
+ }
+
+ rcu_read_lock();
+ lttng_ht_lookup(metadata_ht, (void *)((unsigned long) pollfd),
+ &iter);
+ node = lttng_ht_iter_get_node_ulong(&iter);
+ assert(node);
+
+ stream = caa_container_of(node, struct lttng_consumer_stream,
+ node);
+
+ /* Check for error event */
+ if (revents & (LPOLLERR | LPOLLHUP)) {
+ DBG("Metadata fd %d is hup|err.", pollfd);
+ if (!stream->hangup_flush_done
+ && (consumer_data.type == LTTNG_CONSUMER32_UST
+ || consumer_data.type == LTTNG_CONSUMER64_UST)) {
+ DBG("Attempting to flush and consume the UST buffers");
+ lttng_ustconsumer_on_stream_hangup(stream);
+
+ /* We just flushed the stream now read it. */
+ do {
+ len = ctx->on_buffer_ready(stream, ctx);
+ /*
+ * We don't check the return value here since if we get
+ * a negative len, it means an error occured thus we
+ * simply remove it from the poll set and free the
+ * stream.
+ */
+ } while (len > 0);
+ }
+
+ lttng_poll_del(&events, stream->wait_fd);
+ /*
+ * This call update the channel states, closes file descriptors
+ * and securely free the stream.
+ */
+ consumer_del_metadata_stream(stream, metadata_ht);
+ } else if (revents & (LPOLLIN | LPOLLPRI)) {
+ /* Get the data out of the metadata file descriptor */
+ DBG("Metadata available on fd %d", pollfd);
+ assert(stream->wait_fd == pollfd);
+
+ len = ctx->on_buffer_ready(stream, ctx);
+ /* It's ok to have an unavailable sub-buffer */
+ if (len < 0 && len != -EAGAIN && len != -ENODATA) {
+ /* Clean up stream from consumer and free it. */
+ lttng_poll_del(&events, stream->wait_fd);
+ consumer_del_metadata_stream(stream, metadata_ht);
+ } else if (len > 0) {
+ stream->data_read = 1;
+ }
+ }
+
+ /* Release RCU lock for the stream looked up */
+ rcu_read_unlock();
+ }
+ }
+
+error:
+end:
+ DBG("Metadata poll thread exiting");
+ lttng_poll_clean(&events);
+
+ if (metadata_ht) {
+ destroy_stream_ht(metadata_ht);
+ }
+
+ rcu_unregister_thread();
+ return NULL;
+}
+
+/*
+ * This thread polls the fds in the set to consume the data and write
+ * it to tracefile if necessary.
+ */
+void *consumer_thread_data_poll(void *data)
+{
+ int num_rdy, num_hup, high_prio, ret, i;
+ struct pollfd *pollfd = NULL;
+ /* local view of the streams */
+ struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
+ /* local view of consumer_data.fds_count */
+ int nb_fd = 0;
+ struct lttng_consumer_local_data *ctx = data;
+ ssize_t len;