consumerd: tag metadata channel as being part of a live session
[lttng-tools.git] / src / common / ust-consumer / ust-consumer.c
index 83b2143d727ca5aab4790d665ba229912a1501d6..e5143cd4a5f38afbc8e77189135c68fff25e4e96 100644 (file)
@@ -1,22 +1,13 @@
 /*
- * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
- *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
+ * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
+ * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
  *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License, version 2 only,
- * as published by the Free Software Foundation.
+ * SPDX-License-Identifier: GPL-2.0-only
  *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License along
- * with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
+#include <stdint.h>
 #define _LGPL_SOURCE
 #include <assert.h>
 #include <lttng/ust-ctl.h>
@@ -120,26 +111,6 @@ error:
        return ret;
 }
 
-/*
- * Allocate and return a consumer channel object.
- */
-static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
-               const uint64_t *chunk_id, const char *pathname, const char *name,
-               uint64_t relayd_id, uint64_t key, enum lttng_event_output output,
-               uint64_t tracefile_size, uint64_t tracefile_count,
-               uint64_t session_id_per_pid, unsigned int monitor,
-               unsigned int live_timer_interval,
-               const char *root_shm_path, const char *shm_path)
-{
-       assert(pathname);
-       assert(name);
-
-       return consumer_allocate_channel(key, session_id, chunk_id, pathname,
-                       name, relayd_id, output, tracefile_size,
-                       tracefile_count, session_id_per_pid, monitor,
-                       live_timer_interval, root_shm_path, shm_path);
-}
-
 /*
  * Allocate and return a consumer stream object. If _alloc_ret is not NULL, the
  * error value if applicable is set in it else it is kept untouched.
@@ -156,7 +127,9 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
        assert(channel);
        assert(ctx);
 
-       stream = consumer_allocate_stream(channel->key,
+       stream = consumer_allocate_stream(
+                       channel,
+                       channel->key,
                        key,
                        channel->name,
                        channel->relayd_id,
@@ -185,7 +158,6 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
        }
 
        consumer_stream_update_channel_attributes(stream, channel);
-       stream->chan = channel;
 
 error:
        if (_alloc_ret) {
@@ -519,7 +491,8 @@ error_open:
        if (channel->root_shm_path[0]) {
                (void) run_as_rmdir_recursive(channel->root_shm_path,
                                channel->buffer_credentials.value.uid,
-                               channel->buffer_credentials.value.gid);
+                               channel->buffer_credentials.value.gid,
+                               LTTNG_DIRECTORY_HANDLE_SKIP_NON_EMPTY_FLAG);
        }
        free(stream_fds);
 error_alloc:
@@ -774,10 +747,19 @@ static int flush_channel(uint64_t chan_key)
                health_code_update();
 
                pthread_mutex_lock(&stream->lock);
+
+               /*
+                * Protect against concurrent teardown of a stream.
+                */
+               if (cds_lfht_is_node_deleted(&stream->node.node)) {
+                       goto next;
+               }
+
                if (!stream->quiescent) {
                        ustctl_flush_buffer(stream->ustream, 0);
                        stream->quiescent = true;
                }
+next:
                pthread_mutex_unlock(&stream->lock);
        }
 error:
@@ -1069,7 +1051,6 @@ error_stream:
         * Clean up the stream completly because the next snapshot will use a new
         * metadata stream.
         */
-       pthread_mutex_lock(&metadata_stream->lock);
        consumer_stream_destroy(metadata_stream, NULL);
        cds_list_del(&metadata_stream->send_node);
        metadata_channel->metadata_stream = NULL;
@@ -1079,6 +1060,35 @@ error:
        return ret;
 }
 
+static
+int get_current_subbuf_addr(struct lttng_consumer_stream *stream,
+               const char **addr)
+{
+       int ret;
+       unsigned long mmap_offset;
+       const char *mmap_base;
+
+       mmap_base = ustctl_get_mmap_base(stream->ustream);
+       if (!mmap_base) {
+               ERR("Failed to get mmap base for stream `%s`",
+                               stream->name);
+               ret = -EPERM;
+               goto error;
+       }
+
+       ret = ustctl_get_mmap_read_offset(stream->ustream, &mmap_offset);
+       if (ret != 0) {
+               ERR("Failed to get mmap offset for stream `%s`", stream->name);
+               ret = -EINVAL;
+               goto error;
+       }
+
+       *addr = mmap_base + mmap_offset;
+error:
+       return ret;
+
+}
+
 /*
  * Take a snapshot of all the stream of a channel.
  * RCU read-side lock and the channel lock must be held by the caller.
@@ -1181,6 +1191,8 @@ static int snapshot_channel(struct lttng_consumer_channel *channel,
                while ((long) (consumed_pos - produced_pos) < 0) {
                        ssize_t read_len;
                        unsigned long len, padded_len;
+                       const char *subbuf_addr;
+                       struct lttng_buffer_view subbuf_view;
 
                        health_code_update();
 
@@ -1210,8 +1222,16 @@ static int snapshot_channel(struct lttng_consumer_channel *channel,
                                goto error_put_subbuf;
                        }
 
-                       read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
-                                       padded_len - len, NULL);
+                       ret = get_current_subbuf_addr(stream, &subbuf_addr);
+                       if (ret) {
+                               goto error_put_subbuf;
+                       }
+
+                       subbuf_view = lttng_buffer_view_init(
+                                       subbuf_addr, 0, padded_len);
+                       read_len = lttng_consumer_on_read_subbuffer_mmap(ctx,
+                                       stream, &subbuf_view, padded_len - len,
+                                       NULL);
                        if (use_relayd) {
                                if (read_len != len) {
                                        ret = -EPERM;
@@ -1440,19 +1460,21 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                };
 
                /* Create a plain object and reserve a channel key. */
-               channel = allocate_channel(msg.u.ask_channel.session_id,
+               channel = consumer_allocate_channel(
+                               msg.u.ask_channel.key,
+                               msg.u.ask_channel.session_id,
                                msg.u.ask_channel.chunk_id.is_set ?
                                                &chunk_id : NULL,
                                msg.u.ask_channel.pathname,
                                msg.u.ask_channel.name,
                                msg.u.ask_channel.relayd_id,
-                               msg.u.ask_channel.key,
                                (enum lttng_event_output) msg.u.ask_channel.output,
                                msg.u.ask_channel.tracefile_size,
                                msg.u.ask_channel.tracefile_count,
                                msg.u.ask_channel.session_id_per_pid,
                                msg.u.ask_channel.monitor,
                                msg.u.ask_channel.live_timer_interval,
+                               msg.u.ask_channel.is_live,
                                msg.u.ask_channel.root_shm_path,
                                msg.u.ask_channel.shm_path);
                if (!channel) {
@@ -1590,7 +1612,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                if (!channel) {
                        ERR("UST consumer get channel key %" PRIu64 " not found", key);
                        ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
-                       goto end_msg_sessiond;
+                       goto end_get_channel;
                }
 
                health_code_update();
@@ -1606,13 +1628,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                 * and the consumer can continue its work. The above call
                                 * has sent the error status message to the sessiond.
                                 */
-                               goto end_nosignal;
+                               goto end_get_channel_nosignal;
                        }
                        /*
                         * The communicaton was broken hence there is a bad state between
                         * the consumer and sessiond so stop everything.
                         */
-                       goto error_fatal;
+                       goto error_get_channel_fatal;
                }
 
                health_code_update();
@@ -1622,7 +1644,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                 * so don't send them to the data thread.
                 */
                if (!channel->monitor) {
-                       goto end_msg_sessiond;
+                       goto end_get_channel;
                }
 
                ret = send_streams_to_thread(channel, ctx);
@@ -1631,11 +1653,16 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                         * If we are unable to send the stream to the thread, there is
                         * a big problem so just stop everything.
                         */
-                       goto error_fatal;
+                       goto error_get_channel_fatal;
                }
                /* List MUST be empty after or else it could be reused. */
                assert(cds_list_empty(&channel->streams.head));
+end_get_channel:
                goto end_msg_sessiond;
+error_get_channel_fatal:
+               goto error_fatal;
+end_get_channel_nosignal:
+               goto end_nosignal;
        }
        case LTTNG_CONSUMER_DESTROY_CHANNEL:
        {
@@ -1706,7 +1733,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                         */
                        DBG("UST consumer push metadata %" PRIu64 " not found", key);
                        ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
-                       goto end_msg_sessiond;
+                       goto end_push_metadata_msg_sessiond;
                }
 
                health_code_update();
@@ -1717,14 +1744,14 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                         * checked whether the channel can be found.
                         */
                        ret_code = LTTCOMM_CONSUMERD_SUCCESS;
-                       goto end_msg_sessiond;
+                       goto end_push_metadata_msg_sessiond;
                }
 
                /* Tell session daemon we are ready to receive the metadata. */
                ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
                if (ret < 0) {
                        /* Somehow, the session daemon is not responding anymore. */
-                       goto error_fatal;
+                       goto error_push_metadata_fatal;
                }
 
                health_code_update();
@@ -1734,7 +1761,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                ret = lttng_consumer_poll_socket(consumer_sockpoll);
                health_poll_exit();
                if (ret) {
-                       goto error_fatal;
+                       goto error_push_metadata_fatal;
                }
 
                health_code_update();
@@ -1743,11 +1770,15 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                len, version, channel, 0, 1);
                if (ret < 0) {
                        /* error receiving from sessiond */
-                       goto error_fatal;
+                       goto error_push_metadata_fatal;
                } else {
                        ret_code = ret;
-                       goto end_msg_sessiond;
+                       goto end_push_metadata_msg_sessiond;
                }
+end_push_metadata_msg_sessiond:
+               goto end_msg_sessiond;
+error_push_metadata_fatal:
+               goto error_fatal;
        }
        case LTTNG_CONSUMER_SETUP_METADATA:
        {
@@ -1974,7 +2005,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                ret = consumer_send_status_msg(sock, ret_code);
                if (ret < 0) {
                        /* Somehow, the session daemon is not responding anymore. */
-                       goto end_nosignal;
+                       goto end_rotate_channel_nosignal;
                }
 
                /*
@@ -1992,6 +2023,33 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        }
                }
                break;
+end_rotate_channel_nosignal:
+               goto end_nosignal;
+       }
+       case LTTNG_CONSUMER_CLEAR_CHANNEL:
+       {
+               struct lttng_consumer_channel *channel;
+               uint64_t key = msg.u.clear_channel.key;
+
+               channel = consumer_find_channel(key);
+               if (!channel) {
+                       DBG("Channel %" PRIu64 " not found", key);
+                       ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+               } else {
+                       ret = lttng_consumer_clear_channel(channel);
+                       if (ret) {
+                               ERR("Clear channel failed key %" PRIu64, key);
+                               ret_code = ret;
+                       }
+
+                       health_code_update();
+               }
+               ret = consumer_send_status_msg(sock, ret_code);
+               if (ret < 0) {
+                       /* Somehow, the session daemon is not responding anymore. */
+                       goto end_nosignal;
+               }
+               break;
        }
        case LTTNG_CONSUMER_INIT:
        {
@@ -2008,8 +2066,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
        case LTTNG_CONSUMER_CREATE_TRACE_CHUNK:
        {
                const struct lttng_credentials credentials = {
-                       .uid = msg.u.create_trace_chunk.credentials.uid,
-                       .gid = msg.u.create_trace_chunk.credentials.gid,
+                       .uid = msg.u.create_trace_chunk.credentials.value.uid,
+                       .gid = msg.u.create_trace_chunk.credentials.value.gid,
                };
                const bool is_local_trace =
                                !msg.u.create_trace_chunk.relayd_id.is_set;
@@ -2019,8 +2077,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                *msg.u.create_trace_chunk.override_name ?
                                        msg.u.create_trace_chunk.override_name :
                                        NULL;
-               LTTNG_OPTIONAL(struct lttng_directory_handle) chunk_directory_handle =
-                               LTTNG_OPTIONAL_INIT;
+               struct lttng_directory_handle *chunk_directory_handle = NULL;
 
                /*
                 * The session daemon will only provide a chunk directory file
@@ -2037,56 +2094,74 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                goto end_nosignal;
                        }
 
+                       /*
+                        * Receive trace chunk domain dirfd.
+                        */
                        ret = lttcomm_recv_fds_unix_sock(sock, &chunk_dirfd, 1);
                        if (ret != sizeof(chunk_dirfd)) {
-                               ERR("Failed to receive trace chunk directory file descriptor");
+                               ERR("Failed to receive trace chunk domain directory file descriptor");
                                goto error_fatal;
                        }
 
-                       DBG("Received trace chunk directory fd (%d)",
+                       DBG("Received trace chunk domain directory fd (%d)",
                                        chunk_dirfd);
-                       ret = lttng_directory_handle_init_from_dirfd(
-                                       &chunk_directory_handle.value,
+                       chunk_directory_handle = lttng_directory_handle_create_from_dirfd(
                                        chunk_dirfd);
-                       if (ret) {
-                               ERR("Failed to initialize chunk directory handle from directory file descriptor");
+                       if (!chunk_directory_handle) {
+                               ERR("Failed to initialize chunk domain directory handle from directory file descriptor");
                                if (close(chunk_dirfd)) {
                                        PERROR("Failed to close chunk directory file descriptor");
                                }
                                goto error_fatal;
                        }
-                       chunk_directory_handle.is_set = true;
                }
 
                ret_code = lttng_consumer_create_trace_chunk(
                                !is_local_trace ? &relayd_id : NULL,
                                msg.u.create_trace_chunk.session_id,
                                msg.u.create_trace_chunk.chunk_id,
-                               (time_t) msg.u.create_trace_chunk.creation_timestamp,
+                               (time_t) msg.u.create_trace_chunk
+                                               .creation_timestamp,
                                chunk_override_name,
-                               &credentials,
-                               chunk_directory_handle.is_set ?
-                                               &chunk_directory_handle.value :
-                                               NULL);
-
-               if (chunk_directory_handle.is_set) {
-                       lttng_directory_handle_fini(
-                                       &chunk_directory_handle.value);
-               }
+                               msg.u.create_trace_chunk.credentials.is_set ?
+                                               &credentials :
+                                               NULL,
+                               chunk_directory_handle);
+               lttng_directory_handle_put(chunk_directory_handle);
                goto end_msg_sessiond;
        }
        case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK:
        {
+               enum lttng_trace_chunk_command_type close_command =
+                               msg.u.close_trace_chunk.close_command.value;
                const uint64_t relayd_id =
                                msg.u.close_trace_chunk.relayd_id.value;
+               struct lttcomm_consumer_close_trace_chunk_reply reply;
+               char closed_trace_chunk_path[LTTNG_PATH_MAX];
+               int ret;
 
                ret_code = lttng_consumer_close_trace_chunk(
                                msg.u.close_trace_chunk.relayd_id.is_set ?
-                                               &relayd_id : NULL,
+                                               &relayd_id :
+                                               NULL,
                                msg.u.close_trace_chunk.session_id,
                                msg.u.close_trace_chunk.chunk_id,
-                               (time_t) msg.u.close_trace_chunk.close_timestamp);
-               goto end_msg_sessiond;
+                               (time_t) msg.u.close_trace_chunk.close_timestamp,
+                               msg.u.close_trace_chunk.close_command.is_set ?
+                                               &close_command :
+                                               NULL, closed_trace_chunk_path);
+               reply.ret_code = ret_code;
+               reply.path_length = strlen(closed_trace_chunk_path) + 1;
+               ret = lttcomm_send_unix_sock(sock, &reply, sizeof(reply));
+               if (ret != sizeof(reply)) {
+                       goto error_fatal;
+               }
+               ret = lttcomm_send_unix_sock(sock, closed_trace_chunk_path,
+                               reply.path_length);
+               if (ret != reply.path_length) {
+                       goto error_fatal;
+               }
+               goto end_nosignal;
        }
        case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS:
        {
@@ -2105,15 +2180,12 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
        }
 
 end_nosignal:
-       rcu_read_unlock();
-
-       health_code_update();
-
        /*
         * Return 1 to indicate success since the 0 value can be a socket
         * shutdown during the recv() or send() call.
         */
-       return 1;
+       ret = 1;
+       goto end;
 
 end_msg_sessiond:
        /*
@@ -2125,14 +2197,11 @@ end_msg_sessiond:
        if (ret < 0) {
                goto error_fatal;
        }
-       rcu_read_unlock();
-
-       health_code_update();
+       ret = 1;
+       goto end;
 
-       return 1;
 end_channel_error:
        if (channel) {
-               pthread_mutex_unlock(&channel->lock);
                /*
                 * Free channel here since no one has a reference to it. We don't
                 * free after that because a stream can store this pointer.
@@ -2145,40 +2214,18 @@ end_channel_error:
                /* Stop everything if session daemon can not be notified. */
                goto error_fatal;
        }
-       rcu_read_unlock();
-
-       health_code_update();
+       ret = 1;
+       goto end;
 
-       return 1;
 error_fatal:
-       rcu_read_unlock();
        /* This will issue a consumer stop. */
-       return -1;
-}
-
-/*
- * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
- * compiled out, we isolate it in this library.
- */
-int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
-               unsigned long *off)
-{
-       assert(stream);
-       assert(stream->ustream);
-
-       return ustctl_get_mmap_read_offset(stream->ustream, off);
-}
-
-/*
- * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
- * compiled out, we isolate it in this library.
- */
-void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
-{
-       assert(stream);
-       assert(stream->ustream);
+       ret = -1;
+       goto end;
 
-       return ustctl_get_mmap_base(stream->ustream);
+end:
+       rcu_read_unlock();
+       health_code_update();
+       return ret;
 }
 
 void lttng_ustctl_flush_buffer(struct lttng_consumer_stream *stream,
@@ -2256,6 +2303,14 @@ void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
        ustctl_flush_buffer(stream->ustream, producer);
 }
 
+void lttng_ustconsumer_clear_buffer(struct lttng_consumer_stream *stream)
+{
+       assert(stream);
+       assert(stream->ustream);
+
+       ustctl_clear_buffer(stream->ustream);
+}
+
 int lttng_ustconsumer_get_current_timestamp(
                struct lttng_consumer_stream *stream, uint64_t *ts)
 {
@@ -2340,7 +2395,8 @@ void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan)
        if (chan->root_shm_path[0]) {
                (void) run_as_rmdir_recursive(chan->root_shm_path,
                                chan->buffer_credentials.value.uid,
-                               chan->buffer_credentials.value.gid);
+                               chan->buffer_credentials.value.gid,
+                               LTTNG_DIRECTORY_HANDLE_SKIP_NON_EMPTY_FLAG);
        }
        free(chan->stream_fds);
 }
@@ -2381,62 +2437,69 @@ static int get_index_values(struct ctf_packet_index *index,
                struct ustctl_consumer_stream *ustream)
 {
        int ret;
+       uint64_t packet_size, content_size, timestamp_begin, timestamp_end,
+                       events_discarded, stream_id, stream_instance_id,
+                       packet_seq_num;
 
-       ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
+       ret = ustctl_get_timestamp_begin(ustream, &timestamp_begin);
        if (ret < 0) {
                PERROR("ustctl_get_timestamp_begin");
                goto error;
        }
-       index->timestamp_begin = htobe64(index->timestamp_begin);
 
-       ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
+       ret = ustctl_get_timestamp_end(ustream, &timestamp_end);
        if (ret < 0) {
                PERROR("ustctl_get_timestamp_end");
                goto error;
        }
-       index->timestamp_end = htobe64(index->timestamp_end);
 
-       ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
+       ret = ustctl_get_events_discarded(ustream, &events_discarded);
        if (ret < 0) {
                PERROR("ustctl_get_events_discarded");
                goto error;
        }
-       index->events_discarded = htobe64(index->events_discarded);
 
-       ret = ustctl_get_content_size(ustream, &index->content_size);
+       ret = ustctl_get_content_size(ustream, &content_size);
        if (ret < 0) {
                PERROR("ustctl_get_content_size");
                goto error;
        }
-       index->content_size = htobe64(index->content_size);
 
-       ret = ustctl_get_packet_size(ustream, &index->packet_size);
+       ret = ustctl_get_packet_size(ustream, &packet_size);
        if (ret < 0) {
                PERROR("ustctl_get_packet_size");
                goto error;
        }
-       index->packet_size = htobe64(index->packet_size);
 
-       ret = ustctl_get_stream_id(ustream, &index->stream_id);
+       ret = ustctl_get_stream_id(ustream, &stream_id);
        if (ret < 0) {
                PERROR("ustctl_get_stream_id");
                goto error;
        }
-       index->stream_id = htobe64(index->stream_id);
 
-       ret = ustctl_get_instance_id(ustream, &index->stream_instance_id);
+       ret = ustctl_get_instance_id(ustream, &stream_instance_id);
        if (ret < 0) {
                PERROR("ustctl_get_instance_id");
                goto error;
        }
-       index->stream_instance_id = htobe64(index->stream_instance_id);
 
-       ret = ustctl_get_sequence_number(ustream, &index->packet_seq_num);
+       ret = ustctl_get_sequence_number(ustream, &packet_seq_num);
        if (ret < 0) {
                PERROR("ustctl_get_sequence_number");
                goto error;
        }
-       index->packet_seq_num = htobe64(index->packet_seq_num);
+
+       *index = (typeof(*index)) {
+               .offset = index->offset,
+               .packet_size = htobe64(packet_size),
+               .content_size = htobe64(content_size),
+               .timestamp_begin = htobe64(timestamp_begin),
+               .timestamp_end = htobe64(timestamp_end),
+               .events_discarded = htobe64(events_discarded),
+               .stream_id = htobe64(stream_id),
+               .stream_instance_id = htobe64(stream_instance_id),
+               .packet_seq_num = htobe64(packet_seq_num),
+       };
 
 error:
        return ret;
@@ -2513,6 +2576,13 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
                        stream->ust_metadata_pushed);
        ret = write_len;
 
+       /*
+        * Switch packet (but don't open the next one) on every commit of
+        * a metadata packet. Since the subbuffer is fully filled (with padding,
+        * if needed), the stream is "quiescent" after this commit.
+        */
+       ustctl_flush_buffer(stream->ustream, 1);
+       stream->quiescent = true;
 end:
        pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
        return ret;
@@ -2527,38 +2597,59 @@ end:
  * interacting with sessiond, else we cause a deadlock with live
  * awaiting on metadata to be pushed out.
  *
+ * The RCU read side lock must be held by the caller.
+ *
  * Return 0 if new metadatda is available, EAGAIN if the metadata stream
  * is empty or a negative value on error.
  */
 int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_stream *metadata)
+               struct lttng_consumer_stream *metadata_stream)
 {
        int ret;
        int retry = 0;
+       struct lttng_consumer_channel *metadata_channel;
 
        assert(ctx);
-       assert(metadata);
+       assert(metadata_stream);
 
-       pthread_mutex_unlock(&metadata->lock);
+       metadata_channel = metadata_stream->chan;
+       pthread_mutex_unlock(&metadata_stream->lock);
        /*
         * Request metadata from the sessiond, but don't wait for the flush
         * because we locked the metadata thread.
         */
-       ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0);
-       pthread_mutex_lock(&metadata->lock);
+       ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 0);
+       pthread_mutex_lock(&metadata_stream->lock);
        if (ret < 0) {
                goto end;
        }
 
-       ret = commit_one_metadata_packet(metadata);
+       /*
+        * The metadata stream and channel can be deleted while the
+        * metadata stream lock was released. The streamed is checked
+        * for deletion before we use it further.
+        *
+        * Note that it is safe to access a logically-deleted stream since its
+        * existence is still guaranteed by the RCU read side lock. However,
+        * it should no longer be used. The close/deletion of the metadata
+        * channel and stream already guarantees that all metadata has been
+        * consumed. Therefore, there is nothing left to do in this function.
+        */
+       if (consumer_stream_is_deleted(metadata_stream)) {
+               DBG("Metadata stream %" PRIu64 " was deleted during the metadata synchronization",
+                               metadata_stream->key);
+               ret = 0;
+               goto end;
+       }
+
+       ret = commit_one_metadata_packet(metadata_stream);
        if (ret <= 0) {
                goto end;
        } else if (ret > 0) {
                retry = 1;
        }
 
-       ustctl_flush_buffer(metadata->ustream, 1);
-       ret = ustctl_snapshot(metadata->ustream);
+       ret = ustctl_snapshot(metadata_stream->ustream);
        if (ret < 0) {
                if (errno != EAGAIN) {
                        ERR("Sync metadata, taking UST snapshot");
@@ -2702,6 +2793,8 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
        long ret = 0;
        struct ustctl_consumer_stream *ustream;
        struct ctf_packet_index index;
+       const char *subbuf_addr;
+       struct lttng_buffer_view subbuf_view;
 
        assert(stream);
        assert(stream->ustream);
@@ -2761,7 +2854,6 @@ retry:
                        if (ret <= 0) {
                                goto error;
                        }
-                       ustctl_flush_buffer(stream->ustream, 1);
                        goto retry;
                }
 
@@ -2812,11 +2904,21 @@ retry:
 
        padding = len - subbuf_size;
 
+       ret = get_current_subbuf_addr(stream, &subbuf_addr);
+       if (ret) {
+               write_index = 0;
+               goto error_put_subbuf;
+       }
+
+       subbuf_view = lttng_buffer_view_init(subbuf_addr, 0, len);
+
        /* write the subbuffer to the tracefile */
-       ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
+       ret = lttng_consumer_on_read_subbuffer_mmap(
+                       ctx, stream, &subbuf_view, padding, &index);
        /*
-        * The mmap operation should write subbuf_size amount of data when network
-        * streaming or the full padding (len) size when we are _not_ streaming.
+        * The mmap operation should write subbuf_size amount of data when
+        * network streaming or the full padding (len) size when we are _not_
+        * streaming.
         */
        if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
                        (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
@@ -2833,6 +2935,7 @@ retry:
                                ret, len, subbuf_size);
                write_index = 0;
        }
+error_put_subbuf:
        err = ustctl_put_next_subbuf(ustream);
        assert(err == 0);
 
@@ -3006,7 +3109,7 @@ end:
  * Stop a given metadata channel timer if enabled and close the wait fd which
  * is the poll pipe of the metadata stream.
  *
- * This MUST be called with the metadata channel acquired.
+ * This MUST be called with the metadata channel lock acquired.
  */
 void lttng_ustconsumer_close_metadata(struct lttng_consumer_channel *metadata)
 {
This page took 0.03443 seconds and 4 git commands to generate.