fix: relayd: unaligned access in trace_chunk_registry_ht_key_hash
[lttng-tools.git] / src / bin / lttng-relayd / stream.cpp
index 9df6e52fb52204abfa3bce423503394b72a41b35..a0935231d039280598ad538dfcaaa2f8189d3c27 100644 (file)
@@ -18,6 +18,7 @@
 #include <common/defaults.hpp>
 #include <common/fs-handle.hpp>
 #include <common/sessiond-comm/relayd.hpp>
+#include <common/urcu.hpp>
 #include <common/utils.hpp>
 
 #include <algorithm>
@@ -45,9 +46,9 @@ struct relay_stream *stream_get_by_id(uint64_t stream_id)
 {
        struct lttng_ht_node_u64 *node;
        struct lttng_ht_iter iter;
-       struct relay_stream *stream = NULL;
+       struct relay_stream *stream = nullptr;
 
-       rcu_read_lock();
+       lttng::urcu::read_lock_guard read_lock;
        lttng_ht_lookup(relay_streams_ht, &stream_id, &iter);
        node = lttng_ht_iter_get_node_u64(&iter);
        if (!node) {
@@ -56,10 +57,9 @@ struct relay_stream *stream_get_by_id(uint64_t stream_id)
        }
        stream = lttng::utils::container_of(node, &relay_stream::node);
        if (!stream_get(stream)) {
-               stream = NULL;
+               stream = nullptr;
        }
 end:
-       rcu_read_unlock();
        return stream;
 }
 
@@ -93,7 +93,7 @@ static int stream_create_data_output_file_from_trace_chunk(struct relay_stream *
                                     stream->channel_name,
                                     stream->tracefile_size,
                                     stream->tracefile_current_index,
-                                    NULL,
+                                    nullptr,
                                     stream_path,
                                     sizeof(stream_path));
        if (ret < 0) {
@@ -145,7 +145,7 @@ static int stream_rotate_data_file(struct relay_stream *stream)
 
        if (stream->file) {
                fs_handle_close(stream->file);
-               stream->file = NULL;
+               stream->file = nullptr;
        }
 
        stream->tracefile_wrapped_around = false;
@@ -203,8 +203,8 @@ static int rotate_truncate_stream(struct relay_stream *stream)
        off_t lseek_ret, previous_stream_copy_origin;
        uint64_t copy_bytes_left, misplaced_data_size;
        bool acquired_reference;
-       struct fs_handle *previous_stream_file = NULL;
-       struct lttng_trace_chunk *previous_chunk = NULL;
+       struct fs_handle *previous_stream_file = nullptr;
+       struct lttng_trace_chunk *previous_chunk = nullptr;
 
        if (!LTTNG_OPTIONAL_GET(stream->ongoing_rotation).next_trace_chunk) {
                ERR("Protocol error encoutered in %s(): stream rotation "
@@ -236,7 +236,7 @@ static int rotate_truncate_stream(struct relay_stream *stream)
         */
        LTTNG_ASSERT(stream->file);
        previous_stream_file = stream->file;
-       stream->file = NULL;
+       stream->file = nullptr;
 
        LTTNG_ASSERT(!stream->is_metadata);
        LTTNG_ASSERT(stream->tracefile_size_current > stream->pos_after_last_complete_data_index);
@@ -417,7 +417,7 @@ static int create_index_file(struct relay_stream *stream, struct lttng_trace_chu
 {
        int ret;
        uint32_t major, minor;
-       char *index_subpath = NULL;
+       char *index_subpath = nullptr;
        enum lttng_trace_chunk_status status;
 
        ASSERT_LOCKED(stream->lock);
@@ -425,7 +425,7 @@ static int create_index_file(struct relay_stream *stream, struct lttng_trace_chu
        /* Put ref on previous index_file. */
        if (stream->index_file) {
                lttng_index_file_put(stream->index_file);
-               stream->index_file = NULL;
+               stream->index_file = nullptr;
        }
        major = stream->trace->session->major;
        minor = stream->trace->session->minor;
@@ -516,7 +516,7 @@ static int try_rotate_stream_index(struct relay_stream *stream)
                DBG("Rotating stream %" PRIu64 " index file", stream->stream_handle);
                if (stream->index_file) {
                        lttng_index_file_put(stream->index_file);
-                       stream->index_file = NULL;
+                       stream->index_file = nullptr;
                }
                stream->ongoing_rotation.value.index_rotated = true;
 
@@ -556,7 +556,7 @@ static int stream_set_trace_chunk(struct relay_stream *stream, struct lttng_trac
 
        if (stream->file) {
                fs_handle_close(stream->file);
-               stream->file = NULL;
+               stream->file = nullptr;
        }
        ret = stream_create_data_output_file_from_trace_chunk(stream, chunk, false, &stream->file);
 end:
@@ -574,13 +574,13 @@ struct relay_stream *stream_create(struct ctf_trace *trace,
                                   uint64_t tracefile_count)
 {
        int ret;
-       struct relay_stream *stream = NULL;
+       struct relay_stream *stream = nullptr;
        struct relay_session *session = trace->session;
        bool acquired_reference = false;
        struct lttng_trace_chunk *current_trace_chunk;
 
        stream = zmalloc<relay_stream>();
-       if (stream == NULL) {
+       if (stream == nullptr) {
                PERROR("relay stream zmalloc");
                goto error_no_alloc;
        }
@@ -596,7 +596,7 @@ struct relay_stream *stream_create(struct ctf_trace *trace,
        stream->channel_name = channel_name;
        stream->beacon_ts_end = -1ULL;
        lttng_ht_node_init_u64(&stream->node, stream->stream_handle);
-       pthread_mutex_init(&stream->lock, NULL);
+       pthread_mutex_init(&stream->lock, nullptr);
        urcu_ref_init(&stream->ref);
        ctf_trace_get(trace);
        stream->trace = trace;
@@ -665,10 +665,10 @@ end:
        if (ret) {
                if (stream->file) {
                        fs_handle_close(stream->file);
-                       stream->file = NULL;
+                       stream->file = nullptr;
                }
                stream_put(stream);
-               stream = NULL;
+               stream = nullptr;
        }
        if (acquired_reference) {
                lttng_trace_chunk_put(current_trace_chunk);
@@ -682,7 +682,7 @@ error_no_alloc:
         */
        free(path_name);
        free(channel_name);
-       return NULL;
+       return nullptr;
 }
 
 /*
@@ -788,33 +788,32 @@ static void stream_release(struct urcu_ref *ref)
 
        if (stream->file) {
                fs_handle_close(stream->file);
-               stream->file = NULL;
+               stream->file = nullptr;
        }
        if (stream->index_file) {
                lttng_index_file_put(stream->index_file);
-               stream->index_file = NULL;
+               stream->index_file = nullptr;
        }
        if (stream->trace) {
                ctf_trace_put(stream->trace);
-               stream->trace = NULL;
+               stream->trace = nullptr;
        }
        stream_complete_rotation(stream);
        lttng_trace_chunk_put(stream->trace_chunk);
-       stream->trace_chunk = NULL;
+       stream->trace_chunk = nullptr;
 
        call_rcu(&stream->rcu_node, stream_destroy_rcu);
 }
 
 void stream_put(struct relay_stream *stream)
 {
-       rcu_read_lock();
+       lttng::urcu::read_lock_guard read_lock;
        LTTNG_ASSERT(stream->ref.refcount != 0);
        /*
         * Wait until we have processed all the stream packets before
         * actually putting our last stream reference.
         */
        urcu_ref_put(&stream->ref, stream_release);
-       rcu_read_unlock();
 }
 
 int stream_set_pending_rotation(struct relay_stream *stream,
@@ -972,14 +971,14 @@ void try_stream_close(struct relay_stream *stream)
        /* Put stream fd before put chunk. */
        if (stream->file) {
                fs_handle_close(stream->file);
-               stream->file = NULL;
+               stream->file = nullptr;
        }
        if (stream->index_file) {
                lttng_index_file_put(stream->index_file);
-               stream->index_file = NULL;
+               stream->index_file = nullptr;
        }
        lttng_trace_chunk_put(stream->trace_chunk);
-       stream->trace_chunk = NULL;
+       stream->trace_chunk = nullptr;
        pthread_mutex_unlock(&stream->lock);
        DBG("Succeeded in closing stream %" PRIu64, stream->stream_handle);
        stream_put(stream);
@@ -1029,7 +1028,7 @@ int stream_init_packet(struct relay_stream *stream, size_t packet_size, bool *fi
 
                if (stream->file) {
                        fs_handle_close(stream->file);
-                       stream->file = NULL;
+                       stream->file = nullptr;
                }
                ret = stream_create_data_output_file_from_trace_chunk(
                        stream, stream->trace_chunk, false, &stream->file);
@@ -1110,9 +1109,6 @@ int stream_write(struct relay_stream *stream,
                recv_len = packet ? packet->size : 0;
                recv_len += padding_len;
                stream->metadata_received += recv_len;
-               if (recv_len) {
-                       stream->no_new_metadata_notified = false;
-               }
        }
 
        DBG("Wrote to %sstream %" PRIu64 ": data_length = %zu, padding_length = %zu",
@@ -1169,7 +1165,7 @@ int stream_update_index(struct relay_stream *stream,
                            stream->stream_handle);
                        /* Put self-ref for this index due to error. */
                        relay_index_put(index);
-                       index = NULL;
+                       index = nullptr;
                        goto end;
                }
        }
@@ -1178,7 +1174,7 @@ int stream_update_index(struct relay_stream *stream,
                ret = -1;
                /* Put self-ref for this index due to error. */
                relay_index_put(index);
-               index = NULL;
+               index = nullptr;
                goto end;
        }
 
@@ -1315,18 +1311,20 @@ static void print_stream_indexes(struct relay_stream *stream)
        struct lttng_ht_iter iter;
        struct relay_index *index;
 
-       rcu_read_lock();
-       cds_lfht_for_each_entry (stream->indexes_ht->ht, &iter.iter, index, index_n.node) {
-               DBG("index %p net_seq_num %" PRIu64 " refcount %ld"
-                   " stream %" PRIu64 " trace %" PRIu64 " session %" PRIu64,
-                   index,
-                   index->index_n.key,
-                   stream->ref.refcount,
-                   index->stream->stream_handle,
-                   index->stream->trace->id,
-                   index->stream->trace->session->id);
-       }
-       rcu_read_unlock();
+       {
+               lttng::urcu::read_lock_guard read_lock;
+
+               cds_lfht_for_each_entry (stream->indexes_ht->ht, &iter.iter, index, index_n.node) {
+                       DBG("index %p net_seq_num %" PRIu64 " refcount %ld"
+                           " stream %" PRIu64 " trace %" PRIu64 " session %" PRIu64,
+                           index,
+                           index->index_n.key,
+                           stream->ref.refcount,
+                           index->stream->stream_handle,
+                           index->stream->trace->id,
+                           index->stream->trace->session->id);
+               }
+       }
 }
 
 int stream_reset_file(struct relay_stream *stream)
@@ -1342,7 +1340,7 @@ int stream_reset_file(struct relay_stream *stream)
                            stream->channel_name,
                            stream->stream_handle);
                }
-               stream->file = NULL;
+               stream->file = nullptr;
        }
 
        DBG("%s: reset tracefile_size_current for stream %" PRIu64 " was %" PRIu64,
@@ -1360,7 +1358,7 @@ int stream_reset_file(struct relay_stream *stream)
                stream, stream->trace_chunk, true, &stream->file);
 }
 
-void print_relay_streams(void)
+void print_relay_streams()
 {
        struct lttng_ht_iter iter;
        struct relay_stream *stream;
@@ -1369,19 +1367,23 @@ void print_relay_streams(void)
                return;
        }
 
-       rcu_read_lock();
-       cds_lfht_for_each_entry (relay_streams_ht->ht, &iter.iter, stream, node.node) {
-               if (!stream_get(stream)) {
-                       continue;
+       {
+               lttng::urcu::read_lock_guard read_lock;
+
+               cds_lfht_for_each_entry (relay_streams_ht->ht, &iter.iter, stream, node.node) {
+                       if (!stream_get(stream)) {
+                               continue;
+                       }
+
+                       DBG("stream %p refcount %ld stream %" PRIu64 " trace %" PRIu64
+                           " session %" PRIu64,
+                           stream,
+                           stream->ref.refcount,
+                           stream->stream_handle,
+                           stream->trace->id,
+                           stream->trace->session->id);
+                       print_stream_indexes(stream);
+                       stream_put(stream);
                }
-               DBG("stream %p refcount %ld stream %" PRIu64 " trace %" PRIu64 " session %" PRIu64,
-                   stream,
-                   stream->ref.refcount,
-                   stream->stream_handle,
-                   stream->trace->id,
-                   stream->trace->session->id);
-               print_stream_indexes(stream);
-               stream_put(stream);
        }
-       rcu_read_unlock();
 }
This page took 0.028428 seconds and 4 git commands to generate.