src/bin/lttng-relayd/viewer-stream.cpp

   1 /*
   2  * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
   3  * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
   4  * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * SPDX-License-Identifier: GPL-2.0-only
   7  *
   8  */
   9
  10 #define _LGPL_SOURCE
  11 #include "lttng-relayd.hpp"
  12 #include "viewer-stream.hpp"
  13
  14 #include <common/common.hpp>
  15 #include <common/compat/string.hpp>
  16 #include <common/index/index.hpp>
  17 #include <common/urcu.hpp>
  18 #include <common/utils.hpp>
  19
  20 #include <algorithm>
  21 #include <fcntl.h>
  22 #include <sys/stat.h>
  23 #include <sys/types.h>
  24
  25 static void viewer_stream_release_composite_objects(struct relay_viewer_stream *vstream)
  26 {
  27         if (vstream->stream_file.handle) {
  28                 fs_handle_close(vstream->stream_file.handle);
  29                 vstream->stream_file.handle = nullptr;
  30         }
  31         if (vstream->index_file) {
  32                 lttng_index_file_put(vstream->index_file);
  33                 vstream->index_file = nullptr;
  34         }
  35         if (vstream->stream) {
  36                 stream_put(vstream->stream);
  37                 vstream->stream = nullptr;
  38         }
  39         lttng_trace_chunk_put(vstream->stream_file.trace_chunk);
  40         vstream->stream_file.trace_chunk = nullptr;
  41 }
  42
  43 static void viewer_stream_destroy(struct relay_viewer_stream *vstream)
  44 {
  45         free(vstream->path_name);
  46         free(vstream->channel_name);
  47         free(vstream);
  48 }
  49
  50 static void viewer_stream_destroy_rcu(struct rcu_head *head)
  51 {
  52         struct relay_viewer_stream *vstream =
  53                 lttng::utils::container_of(head, &relay_viewer_stream::rcu_node);
  54
  55         viewer_stream_destroy(vstream);
  56 }
  57
  58 /* Relay stream's lock must be held by the caller. */
  59 struct relay_viewer_stream *viewer_stream_create(struct relay_stream *stream,
  60                                                  struct lttng_trace_chunk *trace_chunk,
  61                                                  enum lttng_viewer_seek seek_t)
  62 {
  63         struct relay_viewer_stream *vstream = nullptr;
  64
  65         ASSERT_LOCKED(stream->lock);
  66
  67         vstream = zmalloc<relay_viewer_stream>();
  68         if (!vstream) {
  69                 PERROR("relay viewer stream zmalloc");
  70                 goto error;
  71         }
  72
  73         if (trace_chunk) {
  74                 const bool acquired_reference = lttng_trace_chunk_get(trace_chunk);
  75
  76                 LTTNG_ASSERT(acquired_reference);
  77         }
  78
  79         vstream->stream_file.trace_chunk = trace_chunk;
  80         vstream->path_name = lttng_strndup(stream->path_name, LTTNG_VIEWER_PATH_MAX);
  81         if (vstream->path_name == nullptr) {
  82                 PERROR("relay viewer path_name alloc");
  83                 goto error;
  84         }
  85         vstream->channel_name = lttng_strndup(stream->channel_name, LTTNG_VIEWER_NAME_MAX);
  86         if (vstream->channel_name == nullptr) {
  87                 PERROR("relay viewer channel_name alloc");
  88                 goto error;
  89         }
  90
  91         if (!stream_get(stream)) {
  92                 ERR("Cannot get stream");
  93                 goto error;
  94         }
  95         vstream->stream = stream;
  96
  97         if (stream->is_metadata && stream->trace->viewer_metadata_stream) {
  98                 ERR("Cannot attach viewer metadata stream to trace (busy).");
  99                 goto error;
 100         }
 101
 102         switch (seek_t) {
 103         case LTTNG_VIEWER_SEEK_BEGINNING:
 104         {
 105                 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
 106
 107                 if (seq_tail == -1ULL) {
 108                         /*
 109                          * Tail may not be initialized yet. Nonetheless, we know
 110                          * we want to send the first index once it becomes
 111                          * available.
 112                          */
 113                         seq_tail = 0;
 114                 }
 115                 vstream->current_tracefile_id = tracefile_array_get_file_index_tail(stream->tfa);
 116                 vstream->index_sent_seqcount = seq_tail;
 117                 break;
 118         }
 119         case LTTNG_VIEWER_SEEK_LAST:
 120                 vstream->current_tracefile_id =
 121                         tracefile_array_get_read_file_index_head(stream->tfa);
 122                 /*
 123                  * We seek at the very end of each stream, awaiting for
 124                  * a future packet to eventually come in.
 125                  *
 126                  * We don't need to check the head position for -1ULL since the
 127                  * increment will set it to 0.
 128                  */
 129                 vstream->index_sent_seqcount = tracefile_array_get_seq_head(stream->tfa) + 1;
 130                 break;
 131         default:
 132                 goto error;
 133         }
 134
 135         /*
 136          * If we never received an index for the current stream, delay
 137          * the opening of the index, otherwise open it right now.
 138          */
 139         if (stream->index_file == nullptr) {
 140                 vstream->index_file = nullptr;
 141         } else if (vstream->stream_file.trace_chunk) {
 142                 const uint32_t connection_major = stream->trace->session->major;
 143                 const uint32_t connection_minor = stream->trace->session->minor;
 144                 enum lttng_trace_chunk_status chunk_status;
 145
 146                 chunk_status = lttng_index_file_create_from_trace_chunk_read_only(
 147                         vstream->stream_file.trace_chunk,
 148                         stream->path_name,
 149                         stream->channel_name,
 150                         stream->tracefile_size,
 151                         vstream->current_tracefile_id,
 152                         lttng_to_index_major(connection_major, connection_minor),
 153                         lttng_to_index_minor(connection_major, connection_minor),
 154                         true,
 155                         &vstream->index_file);
 156                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 157                         if (chunk_status == LTTNG_TRACE_CHUNK_STATUS_NO_FILE) {
 158                                 vstream->index_file = nullptr;
 159                         } else {
 160                                 goto error;
 161                         }
 162                 }
 163         }
 164
 165         /*
 166          * If we never received a data file for the current stream, delay the
 167          * opening, otherwise open it right now.
 168          */
 169         if (stream->file && vstream->stream_file.trace_chunk) {
 170                 int ret;
 171                 char file_path[LTTNG_PATH_MAX];
 172                 enum lttng_trace_chunk_status status;
 173
 174                 ret = utils_stream_file_path(stream->path_name,
 175                                              stream->channel_name,
 176                                              stream->tracefile_size,
 177                                              vstream->current_tracefile_id,
 178                                              nullptr,
 179                                              file_path,
 180                                              sizeof(file_path));
 181                 if (ret < 0) {
 182                         goto error;
 183                 }
 184
 185                 status = lttng_trace_chunk_open_fs_handle(vstream->stream_file.trace_chunk,
 186                                                           file_path,
 187                                                           O_RDONLY,
 188                                                           0,
 189                                                           &vstream->stream_file.handle,
 190                                                           true);
 191                 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 192                         goto error;
 193                 }
 194         }
 195
 196         if (seek_t == LTTNG_VIEWER_SEEK_LAST && vstream->index_file) {
 197                 off_t lseek_ret;
 198
 199                 lseek_ret = fs_handle_seek(vstream->index_file->file, 0, SEEK_END);
 200                 if (lseek_ret < 0) {
 201                         goto error;
 202                 }
 203         }
 204         if (stream->is_metadata) {
 205                 rcu_assign_pointer(stream->trace->viewer_metadata_stream, vstream);
 206         }
 207
 208         vstream->last_seen_rotation_count = stream->completed_rotation_count;
 209
 210         /* Globally visible after the add unique. */
 211         lttng_ht_node_init_u64(&vstream->stream_n, stream->stream_handle);
 212         urcu_ref_init(&vstream->ref);
 213         lttng_ht_add_unique_u64(viewer_streams_ht, &vstream->stream_n);
 214
 215         return vstream;
 216
 217 error:
 218         if (vstream) {
 219                 /* Not using `put` since vstream is assumed to be published. */
 220                 viewer_stream_release_composite_objects(vstream);
 221                 viewer_stream_destroy(vstream);
 222         }
 223         return nullptr;
 224 }
 225
 226 static void viewer_stream_unpublish(struct relay_viewer_stream *vstream)
 227 {
 228         int ret;
 229         struct lttng_ht_iter iter;
 230
 231         iter.iter.node = &vstream->stream_n.node;
 232         ret = lttng_ht_del(viewer_streams_ht, &iter);
 233         LTTNG_ASSERT(!ret);
 234 }
 235
 236 static void viewer_stream_release(struct urcu_ref *ref)
 237 {
 238         struct relay_viewer_stream *vstream =
 239                 caa_container_of(ref, struct relay_viewer_stream, ref);
 240
 241         if (vstream->stream->is_metadata) {
 242                 rcu_assign_pointer(vstream->stream->trace->viewer_metadata_stream, NULL);
 243         }
 244         viewer_stream_unpublish(vstream);
 245         viewer_stream_release_composite_objects(vstream);
 246         call_rcu(&vstream->rcu_node, viewer_stream_destroy_rcu);
 247 }
 248
 249 /* Must be called with RCU read-side lock held. */
 250 bool viewer_stream_get(struct relay_viewer_stream *vstream)
 251 {
 252         return urcu_ref_get_unless_zero(&vstream->ref);
 253 }
 254
 255 /*
 256  * Get viewer stream by id.
 257  *
 258  * Return viewer stream if found else NULL.
 259  */
 260 struct relay_viewer_stream *viewer_stream_get_by_id(uint64_t id)
 261 {
 262         struct lttng_ht_node_u64 *node;
 263         struct lttng_ht_iter iter;
 264         struct relay_viewer_stream *vstream = nullptr;
 265
 266         lttng::urcu::read_lock_guard read_lock;
 267         lttng_ht_lookup(viewer_streams_ht, &id, &iter);
 268         node = lttng_ht_iter_get_node_u64(&iter);
 269         if (!node) {
 270                 DBG("Relay viewer stream %" PRIu64 " not found", id);
 271                 goto end;
 272         }
 273         vstream = lttng::utils::container_of(node, &relay_viewer_stream::stream_n);
 274         if (!viewer_stream_get(vstream)) {
 275                 vstream = nullptr;
 276         }
 277 end:
 278         return vstream;
 279 }
 280
 281 void viewer_stream_put(struct relay_viewer_stream *vstream)
 282 {
 283         lttng::urcu::read_lock_guard read_lock;
 284         urcu_ref_put(&vstream->ref, viewer_stream_release);
 285 }
 286
 287 void viewer_stream_close_files(struct relay_viewer_stream *vstream)
 288 {
 289         if (vstream->index_file) {
 290                 lttng_index_file_put(vstream->index_file);
 291                 vstream->index_file = nullptr;
 292         }
 293         if (vstream->stream_file.handle) {
 294                 fs_handle_close(vstream->stream_file.handle);
 295                 vstream->stream_file.handle = nullptr;
 296         }
 297 }
 298
 299 void viewer_stream_sync_tracefile_array_tail(struct relay_viewer_stream *vstream)
 300 {
 301         const struct relay_stream *stream = vstream->stream;
 302         uint64_t seq_tail;
 303
 304         vstream->current_tracefile_id = tracefile_array_get_file_index_tail(stream->tfa);
 305         seq_tail = tracefile_array_get_seq_tail(stream->tfa);
 306         if (seq_tail == -1ULL) {
 307                 seq_tail = 0;
 308         }
 309
 310         /*
 311          * Move the index sent seqcount forward if it was lagging behind
 312          * the new tail of the tracefile array. If the current
 313          * index_sent_seqcount is already further than the tracefile
 314          * array tail position, keep its current position.
 315          */
 316         vstream->index_sent_seqcount = std::max(seq_tail, vstream->index_sent_seqcount);
 317 }
 318
 319 /*
 320  * Rotate a stream to the next tracefile.
 321  *
 322  * Must be called with the rstream lock held.
 323  * Returns 0 on success, 1 on EOF.
 324  */
 325 int viewer_stream_rotate(struct relay_viewer_stream *vstream)
 326 {
 327         int ret;
 328         uint64_t new_id;
 329         const struct relay_stream *stream = vstream->stream;
 330
 331         /* Detect the last tracefile to open. */
 332         if (stream->index_received_seqcount == vstream->index_sent_seqcount &&
 333             stream->trace->session->connection_closed) {
 334                 ret = 1;
 335                 goto end;
 336         }
 337
 338         if (stream->tracefile_count == 0) {
 339                 /* Ignore rotation, there is none to do. */
 340                 ret = 0;
 341                 goto end;
 342         }
 343
 344         /*
 345          * Try to move to the next file.
 346          */
 347         new_id = (vstream->current_tracefile_id + 1) % stream->tracefile_count;
 348         if (tracefile_array_seq_in_file(stream->tfa, new_id, vstream->index_sent_seqcount)) {
 349                 vstream->current_tracefile_id = new_id;
 350         } else {
 351                 uint64_t seq_tail = tracefile_array_get_seq_tail(stream->tfa);
 352
 353                 /*
 354                  * This can only be reached on overwrite, which implies there
 355                  * has been data written at some point, which will have set the
 356                  * tail.
 357                  */
 358                 LTTNG_ASSERT(seq_tail != -1ULL);
 359                 /*
 360                  * We need to resync because we lag behind tail.
 361                  */
 362                 vstream->current_tracefile_id = tracefile_array_get_file_index_tail(stream->tfa);
 363                 vstream->index_sent_seqcount = seq_tail;
 364         }
 365         viewer_stream_close_files(vstream);
 366         ret = 0;
 367 end:
 368         return ret;
 369 }
 370
 371 void print_viewer_streams()
 372 {
 373         struct lttng_ht_iter iter;
 374         struct relay_viewer_stream *vstream;
 375
 376         if (!viewer_streams_ht) {
 377                 return;
 378         }
 379
 380         {
 381                 lttng::urcu::read_lock_guard read_lock;
 382
 383                 cds_lfht_for_each_entry (
 384                         viewer_streams_ht->ht, &iter.iter, vstream, stream_n.node) {
 385                         if (!viewer_stream_get(vstream)) {
 386                                 continue;
 387                         }
 388                         DBG("vstream %p refcount %ld stream %" PRIu64 " trace %" PRIu64
 389                             " session %" PRIu64,
 390                             vstream,
 391                             vstream->ref.refcount,
 392                             vstream->stream->stream_handle,
 393                             vstream->stream->trace->id,
 394                             vstream->stream->trace->session->id);
 395                         viewer_stream_put(vstream);
 396                 }
 397         }
 398 }