src/common/consumer/consumer-stream.c

   1 /*
   2  * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
   3  * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
   5  *
   6  * SPDX-License-Identifier: GPL-2.0-only
   7  *
   8  */
   9
  10 #define _LGPL_SOURCE
  11 #include <assert.h>
  12 #include <inttypes.h>
  13 #include <sys/mman.h>
  14 #include <unistd.h>
  15
  16 #include <common/common.h>
  17 #include <common/index/index.h>
  18 #include <common/kernel-consumer/kernel-consumer.h>
  19 #include <common/relayd/relayd.h>
  20 #include <common/ust-consumer/ust-consumer.h>
  21 #include <common/utils.h>
  22 #include <common/consumer/consumer.h>
  23 #include <common/consumer/consumer-timer.h>
  24 #include <common/consumer/metadata-bucket.h>
  25 #include <common/kernel-ctl/kernel-ctl.h>
  26
  27 #include "consumer-stream.h"
  28
  29 /*
  30  * RCU call to free stream. MUST only be used with call_rcu().
  31  */
  32 static void free_stream_rcu(struct rcu_head *head)
  33 {
  34         struct lttng_ht_node_u64 *node =
  35                 caa_container_of(head, struct lttng_ht_node_u64, head);
  36         struct lttng_consumer_stream *stream =
  37                 caa_container_of(node, struct lttng_consumer_stream, node);
  38
  39         pthread_mutex_destroy(&stream->lock);
  40         free(stream);
  41 }
  42
  43 static void consumer_stream_data_lock_all(struct lttng_consumer_stream *stream)
  44 {
  45         pthread_mutex_lock(&stream->chan->lock);
  46         pthread_mutex_lock(&stream->lock);
  47 }
  48
  49 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream *stream)
  50 {
  51         pthread_mutex_unlock(&stream->lock);
  52         pthread_mutex_unlock(&stream->chan->lock);
  53 }
  54
  55 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream *stream)
  56 {
  57         consumer_stream_data_lock_all(stream);
  58         pthread_mutex_lock(&stream->metadata_rdv_lock);
  59 }
  60
  61 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream *stream)
  62 {
  63         pthread_mutex_unlock(&stream->metadata_rdv_lock);
  64         consumer_stream_data_unlock_all(stream);
  65 }
  66
  67 /* Only used for data streams. */
  68 static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
  69                 const struct stream_subbuffer *subbuf)
  70 {
  71         int ret = 0;
  72         uint64_t sequence_number;
  73         const uint64_t discarded_events = subbuf->info.data.events_discarded;
  74
  75         if (!subbuf->info.data.sequence_number.is_set) {
  76                 /* Command not supported by the tracer. */
  77                 sequence_number = -1ULL;
  78                 stream->sequence_number_unavailable = true;
  79         } else {
  80                 sequence_number = subbuf->info.data.sequence_number.value;
  81         }
  82
  83         /*
  84          * Start the sequence when we extract the first packet in case we don't
  85          * start at 0 (for example if a consumer is not connected to the
  86          * session immediately after the beginning).
  87          */
  88         if (stream->last_sequence_number == -1ULL) {
  89                 stream->last_sequence_number = sequence_number;
  90         } else if (sequence_number > stream->last_sequence_number) {
  91                 stream->chan->lost_packets += sequence_number -
  92                                 stream->last_sequence_number - 1;
  93         } else {
  94                 /* seq <= last_sequence_number */
  95                 ERR("Sequence number inconsistent : prev = %" PRIu64
  96                     ", current = %" PRIu64,
  97                                 stream->last_sequence_number, sequence_number);
  98                 ret = -1;
  99                 goto end;
 100         }
 101         stream->last_sequence_number = sequence_number;
 102
 103         if (discarded_events < stream->last_discarded_events) {
 104                 /*
 105                  * Overflow has occurred. We assume only one wrap-around
 106                  * has occurred.
 107                  */
 108                 stream->chan->discarded_events +=
 109                                 (1ULL << (CAA_BITS_PER_LONG - 1)) -
 110                                 stream->last_discarded_events +
 111                                 discarded_events;
 112         } else {
 113                 stream->chan->discarded_events += discarded_events -
 114                                                   stream->last_discarded_events;
 115         }
 116         stream->last_discarded_events = discarded_events;
 117         ret = 0;
 118
 119 end:
 120         return ret;
 121 }
 122
 123 static
 124 void ctf_packet_index_populate(struct ctf_packet_index *index,
 125                 off_t offset, const struct stream_subbuffer *subbuffer)
 126 {
 127         *index = (typeof(*index)){
 128                 .offset = htobe64(offset),
 129                 .packet_size = htobe64(subbuffer->info.data.packet_size),
 130                 .content_size = htobe64(subbuffer->info.data.content_size),
 131                 .timestamp_begin = htobe64(
 132                                 subbuffer->info.data.timestamp_begin),
 133                 .timestamp_end = htobe64(
 134                                 subbuffer->info.data.timestamp_end),
 135                 .events_discarded = htobe64(
 136                                 subbuffer->info.data.events_discarded),
 137                 .stream_id = htobe64(subbuffer->info.data.stream_id),
 138                 .stream_instance_id = htobe64(
 139                                 subbuffer->info.data.stream_instance_id.is_set ?
 140                                 subbuffer->info.data.stream_instance_id.value : -1ULL),
 141                 .packet_seq_num = htobe64(
 142                                 subbuffer->info.data.sequence_number.is_set ?
 143                                 subbuffer->info.data.sequence_number.value : -1ULL),
 144         };
 145 }
 146
 147 static ssize_t consumer_stream_consume_mmap(
 148                 struct lttng_consumer_local_data *ctx,
 149                 struct lttng_consumer_stream *stream,
 150                 const struct stream_subbuffer *subbuffer)
 151 {
 152         const unsigned long padding_size =
 153                         subbuffer->info.data.padded_subbuf_size -
 154                         subbuffer->info.data.subbuf_size;
 155
 156         return lttng_consumer_on_read_subbuffer_mmap(
 157                         stream, &subbuffer->buffer.buffer, padding_size);
 158 }
 159
 160 static ssize_t consumer_stream_consume_splice(
 161                 struct lttng_consumer_local_data *ctx,
 162                 struct lttng_consumer_stream *stream,
 163                 const struct stream_subbuffer *subbuffer)
 164 {
 165         return lttng_consumer_on_read_subbuffer_splice(ctx, stream,
 166                         subbuffer->info.data.padded_subbuf_size, 0);
 167 }
 168
 169 static int consumer_stream_send_index(
 170                 struct lttng_consumer_stream *stream,
 171                 const struct stream_subbuffer *subbuffer,
 172                 struct lttng_consumer_local_data *ctx)
 173 {
 174         off_t packet_offset = 0;
 175         struct ctf_packet_index index = {};
 176
 177         /*
 178          * This is called after consuming the sub-buffer; substract the
 179          * effect this sub-buffer from the offset.
 180          */
 181         if (stream->net_seq_idx == (uint64_t) -1ULL) {
 182                 packet_offset = stream->out_fd_offset -
 183                                 subbuffer->info.data.padded_subbuf_size;
 184         }
 185
 186         ctf_packet_index_populate(&index, packet_offset, subbuffer);
 187         return consumer_stream_write_index(stream, &index);
 188 }
 189
 190 /*
 191  * Actually do the metadata sync using the given metadata stream.
 192  *
 193  * Return 0 on success else a negative value. ENODATA can be returned also
 194  * indicating that there is no metadata available for that stream.
 195  */
 196 static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 197                 struct lttng_consumer_local_data *ctx)
 198 {
 199         int ret;
 200         enum sync_metadata_status status;
 201
 202         assert(metadata);
 203         assert(metadata->metadata_flag);
 204         assert(ctx);
 205
 206         /*
 207          * In UST, since we have to write the metadata from the cache packet
 208          * by packet, we might need to start this procedure multiple times
 209          * until all the metadata from the cache has been extracted.
 210          */
 211         do {
 212                 /*
 213                  * Steps :
 214                  * - Lock the metadata stream
 215                  * - Check if metadata stream node was deleted before locking.
 216                  *   - if yes, release and return success
 217                  * - Check if new metadata is ready (flush + snapshot pos)
 218                  * - If nothing : release and return.
 219                  * - Lock the metadata_rdv_lock
 220                  * - Unlock the metadata stream
 221                  * - cond_wait on metadata_rdv to wait the wakeup from the
 222                  *   metadata thread
 223                  * - Unlock the metadata_rdv_lock
 224                  */
 225                 pthread_mutex_lock(&metadata->lock);
 226
 227                 /*
 228                  * There is a possibility that we were able to acquire a reference on the
 229                  * stream from the RCU hash table but between then and now, the node might
 230                  * have been deleted just before the lock is acquired. Thus, after locking,
 231                  * we make sure the metadata node has not been deleted which means that the
 232                  * buffers are closed.
 233                  *
 234                  * In that case, there is no need to sync the metadata hence returning a
 235                  * success return code.
 236                  */
 237                 ret = cds_lfht_is_node_deleted(&metadata->node.node);
 238                 if (ret) {
 239                         ret = 0;
 240                         goto end_unlock_mutex;
 241                 }
 242
 243                 switch (ctx->type) {
 244                 case LTTNG_CONSUMER_KERNEL:
 245                         /*
 246                          * Empty the metadata cache and flush the current stream.
 247                          */
 248                         status = lttng_kconsumer_sync_metadata(metadata);
 249                         break;
 250                 case LTTNG_CONSUMER32_UST:
 251                 case LTTNG_CONSUMER64_UST:
 252                         /*
 253                          * Ask the sessiond if we have new metadata waiting and update the
 254                          * consumer metadata cache.
 255                          */
 256                         status = lttng_ustconsumer_sync_metadata(ctx, metadata);
 257                         break;
 258                 default:
 259                         abort();
 260                 }
 261
 262                 switch (status) {
 263                 case SYNC_METADATA_STATUS_NEW_DATA:
 264                         break;
 265                 case SYNC_METADATA_STATUS_NO_DATA:
 266                         ret = 0;
 267                         goto end_unlock_mutex;
 268                 case SYNC_METADATA_STATUS_ERROR:
 269                         ret = -1;
 270                         goto end_unlock_mutex;
 271                 default:
 272                         abort();
 273                 }
 274
 275                 /*
 276                  * At this point, new metadata have been flushed, so we wait on the
 277                  * rendez-vous point for the metadata thread to wake us up when it
 278                  * finishes consuming the metadata and continue execution.
 279                  */
 280
 281                 pthread_mutex_lock(&metadata->metadata_rdv_lock);
 282
 283                 /*
 284                  * Release metadata stream lock so the metadata thread can process it.
 285                  */
 286                 pthread_mutex_unlock(&metadata->lock);
 287
 288                 /*
 289                  * Wait on the rendez-vous point. Once woken up, it means the metadata was
 290                  * consumed and thus synchronization is achieved.
 291                  */
 292                 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
 293                 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
 294         } while (status == SYNC_METADATA_STATUS_NEW_DATA);
 295
 296         /* Success */
 297         return 0;
 298
 299 end_unlock_mutex:
 300         pthread_mutex_unlock(&metadata->lock);
 301         return ret;
 302 }
 303
 304 /*
 305  * Synchronize the metadata using a given session ID. A successful acquisition
 306  * of a metadata stream will trigger a request to the session daemon and a
 307  * snapshot so the metadata thread can consume it.
 308  *
 309  * This function call is a rendez-vous point between the metadata thread and
 310  * the data thread.
 311  *
 312  * Return 0 on success or else a negative value.
 313  */
 314 int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
 315                 uint64_t session_id)
 316 {
 317         int ret;
 318         struct lttng_consumer_stream *stream = NULL;
 319         struct lttng_ht_iter iter;
 320         struct lttng_ht *ht;
 321
 322         assert(ctx);
 323
 324         /* Ease our life a bit. */
 325         ht = consumer_data.stream_list_ht;
 326
 327         rcu_read_lock();
 328
 329         /* Search the metadata associated with the session id of the given stream. */
 330
 331         cds_lfht_for_each_entry_duplicate(ht->ht,
 332                         ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
 333                         &session_id, &iter.iter, stream, node_session_id.node) {
 334                 if (!stream->metadata_flag) {
 335                         continue;
 336                 }
 337
 338                 ret = do_sync_metadata(stream, ctx);
 339                 if (ret < 0) {
 340                         goto end;
 341                 }
 342         }
 343
 344         /*
 345          * Force return code to 0 (success) since ret might be ENODATA for instance
 346          * which is not an error but rather that we should come back.
 347          */
 348         ret = 0;
 349
 350 end:
 351         rcu_read_unlock();
 352         return ret;
 353 }
 354
 355 static int consumer_stream_sync_metadata_index(
 356                 struct lttng_consumer_stream *stream,
 357                 const struct stream_subbuffer *subbuffer,
 358                 struct lttng_consumer_local_data *ctx)
 359 {
 360         int ret;
 361
 362         /* Block until all the metadata is sent. */
 363         pthread_mutex_lock(&stream->metadata_timer_lock);
 364         assert(!stream->missed_metadata_flush);
 365         stream->waiting_on_metadata = true;
 366         pthread_mutex_unlock(&stream->metadata_timer_lock);
 367
 368         ret = consumer_stream_sync_metadata(ctx, stream->session_id);
 369
 370         pthread_mutex_lock(&stream->metadata_timer_lock);
 371         stream->waiting_on_metadata = false;
 372         if (stream->missed_metadata_flush) {
 373                 stream->missed_metadata_flush = false;
 374                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 375                 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
 376         } else {
 377                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 378         }
 379         if (ret < 0) {
 380                 goto end;
 381         }
 382
 383         ret = consumer_stream_send_index(stream, subbuffer, ctx);
 384 end:
 385         return ret;
 386 }
 387
 388 /*
 389  * Check if the local version of the metadata stream matches with the version
 390  * of the metadata stream in the kernel. If it was updated, set the reset flag
 391  * on the stream.
 392  */
 393 static
 394 int metadata_stream_check_version(struct lttng_consumer_stream *stream,
 395         const struct stream_subbuffer *subbuffer)
 396 {
 397         if (stream->metadata_version == subbuffer->info.metadata.version) {
 398                 goto end;
 399         }
 400
 401         DBG("New metadata version detected");
 402         consumer_stream_metadata_set_version(stream,
 403                         subbuffer->info.metadata.version);
 404
 405         if (stream->read_subbuffer_ops.reset_metadata) {
 406                 stream->read_subbuffer_ops.reset_metadata(stream);
 407         }
 408
 409 end:
 410         return 0;
 411 }
 412
 413 static
 414 bool stream_is_rotating_to_null_chunk(
 415                 const struct lttng_consumer_stream *stream)
 416 {
 417         bool rotating_to_null_chunk = false;
 418
 419         if (stream->rotate_position == -1ULL) {
 420                 /* No rotation ongoing. */
 421                 goto end;
 422         }
 423
 424         if (stream->trace_chunk == stream->chan->trace_chunk ||
 425                         !stream->chan->trace_chunk) {
 426                 rotating_to_null_chunk = true;
 427         }
 428 end:
 429         return rotating_to_null_chunk;
 430 }
 431
 432 enum consumer_stream_open_packet_status consumer_stream_open_packet(
 433                 struct lttng_consumer_stream *stream)
 434 {
 435         int ret;
 436         enum consumer_stream_open_packet_status status;
 437         unsigned long produced_pos_before, produced_pos_after;
 438
 439         ret = lttng_consumer_sample_snapshot_positions(stream);
 440         if (ret < 0) {
 441                 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
 442                                 ", channel name = %s, session id = %" PRIu64,
 443                                 stream->key, stream->chan->name,
 444                                 stream->chan->session_id);
 445                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 446                 goto end;
 447         }
 448
 449         ret = lttng_consumer_get_produced_snapshot(
 450                         stream, &produced_pos_before);
 451         if (ret < 0) {
 452                 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
 453                                 ", channel name = %s, session id = %" PRIu64,
 454                                 stream->key, stream->chan->name,
 455                                 stream->chan->session_id);
 456                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 457                 goto end;
 458         }
 459
 460         ret = consumer_stream_flush_buffer(stream, 0);
 461         if (ret) {
 462                 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
 463                                 ", channel name = %s, session id = %" PRIu64,
 464                                 stream->key, stream->chan->name,
 465                                 stream->chan->session_id);
 466                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 467                 goto end;
 468         }
 469
 470         ret = lttng_consumer_sample_snapshot_positions(stream);
 471         if (ret < 0) {
 472                 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
 473                                 ", channel name = %s, session id = %" PRIu64,
 474                                 stream->key, stream->chan->name,
 475                                 stream->chan->session_id);
 476                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 477                 goto end;
 478         }
 479
 480         ret = lttng_consumer_get_produced_snapshot(stream, &produced_pos_after);
 481         if (ret < 0) {
 482                 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
 483                                 ", channel name = %s, session id = %" PRIu64,
 484                                 stream->key, stream->chan->name,
 485                                 stream->chan->session_id);
 486                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 487                 goto end;
 488         }
 489
 490         /*
 491          * Determine if the flush had an effect by comparing the produced
 492          * positons before and after the flush.
 493          */
 494         status = produced_pos_before != produced_pos_after ?
 495                         CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED :
 496                         CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE;
 497         if (status == CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED) {
 498                 stream->opened_packet_in_current_trace_chunk = true;
 499         }
 500
 501 end:
 502         return status;
 503 }
 504
 505 /*
 506  * An attempt to open a new packet is performed after a rotation completes to
 507  * get a begin timestamp as close as possible to the rotation point.
 508  *
 509  * However, that initial attempt at opening a packet can fail due to a full
 510  * ring-buffer. In that case, a second attempt is performed after consuming
 511  * a packet since that will have freed enough space in the ring-buffer.
 512  */
 513 static
 514 int post_consume_open_new_packet(struct lttng_consumer_stream *stream,
 515                 const struct stream_subbuffer *subbuffer,
 516                 struct lttng_consumer_local_data *ctx)
 517 {
 518         int ret = 0;
 519
 520         if (!stream->opened_packet_in_current_trace_chunk &&
 521                         stream->trace_chunk &&
 522                         !stream_is_rotating_to_null_chunk(stream)) {
 523                 const enum consumer_stream_open_packet_status status =
 524                                 consumer_stream_open_packet(stream);
 525
 526                 switch (status) {
 527                 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
 528                         DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
 529                                         ", channel name = %s, session id = %" PRIu64,
 530                                         stream->key, stream->chan->name,
 531                                         stream->chan->session_id);
 532                         stream->opened_packet_in_current_trace_chunk = true;
 533                         break;
 534                 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
 535                         /*
 536                          * Can't open a packet as there is no space left.
 537                          * This means that new events were produced, resulting
 538                          * in a packet being opened, which is what we want
 539                          * anyhow.
 540                          */
 541                         DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
 542                                         ", channel name = %s, session id = %" PRIu64,
 543                                         stream->key, stream->chan->name,
 544                                         stream->chan->session_id);
 545                         stream->opened_packet_in_current_trace_chunk = true;
 546                         break;
 547                 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
 548                         /* Logged by callee. */
 549                         ret = -1;
 550                         goto end;
 551                 default:
 552                         abort();
 553                 }
 554
 555                 stream->opened_packet_in_current_trace_chunk = true;
 556         }
 557
 558 end:
 559         return ret;
 560 }
 561
 562 struct lttng_consumer_stream *consumer_stream_create(
 563                 struct lttng_consumer_channel *channel,
 564                 uint64_t channel_key,
 565                 uint64_t stream_key,
 566                 const char *channel_name,
 567                 uint64_t relayd_id,
 568                 uint64_t session_id,
 569                 struct lttng_trace_chunk *trace_chunk,
 570                 int cpu,
 571                 int *alloc_ret,
 572                 enum consumer_channel_type type,
 573                 unsigned int monitor)
 574 {
 575         int ret;
 576         struct lttng_consumer_stream *stream;
 577
 578         stream = zmalloc(sizeof(*stream));
 579         if (stream == NULL) {
 580                 PERROR("malloc struct lttng_consumer_stream");
 581                 ret = -ENOMEM;
 582                 goto end;
 583         }
 584
 585         if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
 586                 ERR("Failed to acquire trace chunk reference during the creation of a stream");
 587                 ret = -1;
 588                 goto error;
 589         }
 590
 591         rcu_read_lock();
 592         stream->chan = channel;
 593         stream->key = stream_key;
 594         stream->trace_chunk = trace_chunk;
 595         stream->out_fd = -1;
 596         stream->out_fd_offset = 0;
 597         stream->output_written = 0;
 598         stream->net_seq_idx = relayd_id;
 599         stream->session_id = session_id;
 600         stream->monitor = monitor;
 601         stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
 602         stream->index_file = NULL;
 603         stream->last_sequence_number = -1ULL;
 604         stream->rotate_position = -1ULL;
 605         /* Buffer is created with an open packet. */
 606         stream->opened_packet_in_current_trace_chunk = true;
 607         pthread_mutex_init(&stream->lock, NULL);
 608         pthread_mutex_init(&stream->metadata_timer_lock, NULL);
 609
 610         /* If channel is the metadata, flag this stream as metadata. */
 611         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 612                 stream->metadata_flag = 1;
 613                 /* Metadata is flat out. */
 614                 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
 615                 /* Live rendez-vous point. */
 616                 pthread_cond_init(&stream->metadata_rdv, NULL);
 617                 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
 618         } else {
 619                 /* Format stream name to <channel_name>_<cpu_number> */
 620                 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
 621                                 channel_name, cpu);
 622                 if (ret < 0) {
 623                         PERROR("snprintf stream name");
 624                         goto error;
 625                 }
 626         }
 627
 628         switch (channel->output) {
 629         case CONSUMER_CHANNEL_SPLICE:
 630                 stream->output = LTTNG_EVENT_SPLICE;
 631                 ret = utils_create_pipe(stream->splice_pipe);
 632                 if (ret < 0) {
 633                         goto error;
 634                 }
 635                 break;
 636         case CONSUMER_CHANNEL_MMAP:
 637                 stream->output = LTTNG_EVENT_MMAP;
 638                 break;
 639         default:
 640                 abort();
 641         }
 642
 643         /* Key is always the wait_fd for streams. */
 644         lttng_ht_node_init_u64(&stream->node, stream->key);
 645
 646         /* Init node per channel id key */
 647         lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
 648
 649         /* Init session id node with the stream session id */
 650         lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
 651
 652         DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
 653                         " relayd_id %" PRIu64 ", session_id %" PRIu64,
 654                         stream->name, stream->key, channel_key,
 655                         stream->net_seq_idx, stream->session_id);
 656
 657         rcu_read_unlock();
 658
 659         lttng_dynamic_array_init(&stream->read_subbuffer_ops.post_consume_cbs,
 660                         sizeof(post_consume_cb), NULL);
 661
 662         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 663                 stream->read_subbuffer_ops.lock =
 664                                 consumer_stream_metadata_lock_all;
 665                 stream->read_subbuffer_ops.unlock =
 666                                 consumer_stream_metadata_unlock_all;
 667                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 668                                 metadata_stream_check_version;
 669         } else {
 670                 const post_consume_cb post_consume_index_op = channel->is_live ?
 671                                 consumer_stream_sync_metadata_index :
 672                                 consumer_stream_send_index;
 673
 674                 ret = lttng_dynamic_array_add_element(
 675                                 &stream->read_subbuffer_ops.post_consume_cbs,
 676                                 &post_consume_index_op);
 677                 if (ret) {
 678                         PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
 679                         goto error;
 680                 }
 681
 682                 ret = lttng_dynamic_array_add_element(
 683                                 &stream->read_subbuffer_ops.post_consume_cbs,
 684                                 &(post_consume_cb) { post_consume_open_new_packet });
 685                 if (ret) {
 686                         PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
 687                         goto error;
 688                 }
 689
 690                 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
 691                 stream->read_subbuffer_ops.unlock =
 692                                 consumer_stream_data_unlock_all;
 693                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 694                                 consumer_stream_update_stats;
 695         }
 696
 697         if (channel->output == CONSUMER_CHANNEL_MMAP) {
 698                 stream->read_subbuffer_ops.consume_subbuffer =
 699                                 consumer_stream_consume_mmap;
 700         } else {
 701                 stream->read_subbuffer_ops.consume_subbuffer =
 702                                 consumer_stream_consume_splice;
 703         }
 704
 705         return stream;
 706
 707 error:
 708         rcu_read_unlock();
 709         lttng_trace_chunk_put(stream->trace_chunk);
 710         lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
 711         free(stream);
 712 end:
 713         if (alloc_ret) {
 714                 *alloc_ret = ret;
 715         }
 716         return NULL;
 717 }
 718
 719 /*
 720  * Close stream on the relayd side. This call can destroy a relayd if the
 721  * conditions are met.
 722  *
 723  * A RCU read side lock MUST be acquired if the relayd object was looked up in
 724  * a hash table before calling this.
 725  */
 726 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
 727                 struct consumer_relayd_sock_pair *relayd)
 728 {
 729         int ret;
 730
 731         assert(stream);
 732         assert(relayd);
 733
 734         if (stream->sent_to_relayd) {
 735                 uatomic_dec(&relayd->refcount);
 736                 assert(uatomic_read(&relayd->refcount) >= 0);
 737         }
 738
 739         /* Closing streams requires to lock the control socket. */
 740         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 741         ret = relayd_send_close_stream(&relayd->control_sock,
 742                         stream->relayd_stream_id,
 743                         stream->next_net_seq_num - 1);
 744         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 745         if (ret < 0) {
 746                 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
 747                 lttng_consumer_cleanup_relayd(relayd);
 748         }
 749
 750         /* Both conditions are met, we destroy the relayd. */
 751         if (uatomic_read(&relayd->refcount) == 0 &&
 752                         uatomic_read(&relayd->destroy_flag)) {
 753                 consumer_destroy_relayd(relayd);
 754         }
 755         stream->net_seq_idx = (uint64_t) -1ULL;
 756         stream->sent_to_relayd = 0;
 757 }
 758
 759 /*
 760  * Close stream's file descriptors and, if needed, close stream also on the
 761  * relayd side.
 762  *
 763  * The consumer data lock MUST be acquired.
 764  * The stream lock MUST be acquired.
 765  */
 766 void consumer_stream_close(struct lttng_consumer_stream *stream)
 767 {
 768         int ret;
 769         struct consumer_relayd_sock_pair *relayd;
 770
 771         assert(stream);
 772
 773         switch (consumer_data.type) {
 774         case LTTNG_CONSUMER_KERNEL:
 775                 if (stream->mmap_base != NULL) {
 776                         ret = munmap(stream->mmap_base, stream->mmap_len);
 777                         if (ret != 0) {
 778                                 PERROR("munmap");
 779                         }
 780                 }
 781
 782                 if (stream->wait_fd >= 0) {
 783                         ret = close(stream->wait_fd);
 784                         if (ret) {
 785                                 PERROR("close");
 786                         }
 787                         stream->wait_fd = -1;
 788                 }
 789                 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
 790                         utils_close_pipe(stream->splice_pipe);
 791                 }
 792                 break;
 793         case LTTNG_CONSUMER32_UST:
 794         case LTTNG_CONSUMER64_UST:
 795         {
 796                 /*
 797                  * Special case for the metadata since the wait fd is an internal pipe
 798                  * polled in the metadata thread.
 799                  */
 800                 if (stream->metadata_flag && stream->chan->monitor) {
 801                         int rpipe = stream->ust_metadata_poll_pipe[0];
 802
 803                         /*
 804                          * This will stop the channel timer if one and close the write side
 805                          * of the metadata poll pipe.
 806                          */
 807                         lttng_ustconsumer_close_metadata(stream->chan);
 808                         if (rpipe >= 0) {
 809                                 ret = close(rpipe);
 810                                 if (ret < 0) {
 811                                         PERROR("closing metadata pipe read side");
 812                                 }
 813                                 stream->ust_metadata_poll_pipe[0] = -1;
 814                         }
 815                 }
 816                 break;
 817         }
 818         default:
 819                 ERR("Unknown consumer_data type");
 820                 assert(0);
 821         }
 822
 823         /* Close output fd. Could be a socket or local file at this point. */
 824         if (stream->out_fd >= 0) {
 825                 ret = close(stream->out_fd);
 826                 if (ret) {
 827                         PERROR("close");
 828                 }
 829                 stream->out_fd = -1;
 830         }
 831
 832         if (stream->index_file) {
 833                 lttng_index_file_put(stream->index_file);
 834                 stream->index_file = NULL;
 835         }
 836
 837         lttng_trace_chunk_put(stream->trace_chunk);
 838         stream->trace_chunk = NULL;
 839
 840         /* Check and cleanup relayd if needed. */
 841         rcu_read_lock();
 842         relayd = consumer_find_relayd(stream->net_seq_idx);
 843         if (relayd != NULL) {
 844                 consumer_stream_relayd_close(stream, relayd);
 845         }
 846         rcu_read_unlock();
 847 }
 848
 849 /*
 850  * Delete the stream from all possible hash tables.
 851  *
 852  * The consumer data lock MUST be acquired.
 853  * The stream lock MUST be acquired.
 854  */
 855 void consumer_stream_delete(struct lttng_consumer_stream *stream,
 856                 struct lttng_ht *ht)
 857 {
 858         int ret;
 859         struct lttng_ht_iter iter;
 860
 861         assert(stream);
 862         /* Should NEVER be called not in monitor mode. */
 863         assert(stream->chan->monitor);
 864
 865         rcu_read_lock();
 866
 867         if (ht) {
 868                 iter.iter.node = &stream->node.node;
 869                 ret = lttng_ht_del(ht, &iter);
 870                 assert(!ret);
 871         }
 872
 873         /* Delete from stream per channel ID hash table. */
 874         iter.iter.node = &stream->node_channel_id.node;
 875         /*
 876          * The returned value is of no importance. Even if the node is NOT in the
 877          * hash table, we continue since we may have been called by a code path
 878          * that did not add the stream to a (all) hash table. Same goes for the
 879          * next call ht del call.
 880          */
 881         (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
 882
 883         /* Delete from the global stream list. */
 884         iter.iter.node = &stream->node_session_id.node;
 885         /* See the previous ht del on why we ignore the returned value. */
 886         (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
 887
 888         rcu_read_unlock();
 889
 890         if (!stream->metadata_flag) {
 891                 /* Decrement the stream count of the global consumer data. */
 892                 assert(consumer_data.stream_count > 0);
 893                 consumer_data.stream_count--;
 894         }
 895 }
 896
 897 /*
 898  * Free the given stream within a RCU call.
 899  */
 900 void consumer_stream_free(struct lttng_consumer_stream *stream)
 901 {
 902         assert(stream);
 903
 904         metadata_bucket_destroy(stream->metadata_bucket);
 905         call_rcu(&stream->node.head, free_stream_rcu);
 906 }
 907
 908 /*
 909  * Destroy the stream's buffers of the tracer.
 910  */
 911 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 912 {
 913         assert(stream);
 914
 915         switch (consumer_data.type) {
 916         case LTTNG_CONSUMER_KERNEL:
 917                 break;
 918         case LTTNG_CONSUMER32_UST:
 919         case LTTNG_CONSUMER64_UST:
 920                 lttng_ustconsumer_del_stream(stream);
 921                 break;
 922         default:
 923                 ERR("Unknown consumer_data type");
 924                 assert(0);
 925         }
 926 }
 927
 928 /*
 929  * Destroy and close a already created stream.
 930  */
 931 static void destroy_close_stream(struct lttng_consumer_stream *stream)
 932 {
 933         assert(stream);
 934
 935         DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
 936
 937         /* Destroy tracer buffers of the stream. */
 938         consumer_stream_destroy_buffers(stream);
 939         /* Close down everything including the relayd if one. */
 940         consumer_stream_close(stream);
 941 }
 942
 943 /*
 944  * Decrement the stream's channel refcount and if down to 0, return the channel
 945  * pointer so it can be destroyed by the caller or NULL if not.
 946  */
 947 static struct lttng_consumer_channel *unref_channel(
 948                 struct lttng_consumer_stream *stream)
 949 {
 950         struct lttng_consumer_channel *free_chan = NULL;
 951
 952         assert(stream);
 953         assert(stream->chan);
 954
 955         /* Update refcount of channel and see if we need to destroy it. */
 956         if (!uatomic_sub_return(&stream->chan->refcount, 1)
 957                         && !uatomic_read(&stream->chan->nb_init_stream_left)) {
 958                 free_chan = stream->chan;
 959         }
 960
 961         return free_chan;
 962 }
 963
 964 /*
 965  * Destroy a stream completely. This will delete, close and free the stream.
 966  * Once return, the stream is NO longer usable. Its channel may get destroyed
 967  * if conditions are met for a monitored stream.
 968  *
 969  * This MUST be called WITHOUT the consumer data and stream lock acquired if
 970  * the stream is in _monitor_ mode else it does not matter.
 971  */
 972 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 973                 struct lttng_ht *ht)
 974 {
 975         assert(stream);
 976
 977         /* Stream is in monitor mode. */
 978         if (stream->monitor) {
 979                 struct lttng_consumer_channel *free_chan = NULL;
 980
 981                 /*
 982                  * This means that the stream was successfully removed from the streams
 983                  * list of the channel and sent to the right thread managing this
 984                  * stream thus being globally visible.
 985                  */
 986                 if (stream->globally_visible) {
 987                         pthread_mutex_lock(&consumer_data.lock);
 988                         pthread_mutex_lock(&stream->chan->lock);
 989                         pthread_mutex_lock(&stream->lock);
 990                         /* Remove every reference of the stream in the consumer. */
 991                         consumer_stream_delete(stream, ht);
 992
 993                         destroy_close_stream(stream);
 994
 995                         /* Update channel's refcount of the stream. */
 996                         free_chan = unref_channel(stream);
 997
 998                         /* Indicates that the consumer data state MUST be updated after this. */
 999                         consumer_data.need_update = 1;
1000
1001                         pthread_mutex_unlock(&stream->lock);
1002                         pthread_mutex_unlock(&stream->chan->lock);
1003                         pthread_mutex_unlock(&consumer_data.lock);
1004                 } else {
1005                         /*
1006                          * If the stream is not visible globally, this needs to be done
1007                          * outside of the consumer data lock section.
1008                          */
1009                         free_chan = unref_channel(stream);
1010                 }
1011
1012                 if (free_chan) {
1013                         consumer_del_channel(free_chan);
1014                 }
1015         } else {
1016                 destroy_close_stream(stream);
1017         }
1018
1019         /* Free stream within a RCU call. */
1020         lttng_trace_chunk_put(stream->trace_chunk);
1021         stream->trace_chunk = NULL;
1022         lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
1023         consumer_stream_free(stream);
1024 }
1025
1026 /*
1027  * Write index of a specific stream either on the relayd or local disk.
1028  *
1029  * Return 0 on success or else a negative value.
1030  */
1031 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
1032                 struct ctf_packet_index *element)
1033 {
1034         int ret;
1035
1036         assert(stream);
1037         assert(element);
1038
1039         rcu_read_lock();
1040         if (stream->net_seq_idx != (uint64_t) -1ULL) {
1041                 struct consumer_relayd_sock_pair *relayd;
1042                 relayd = consumer_find_relayd(stream->net_seq_idx);
1043                 if (relayd) {
1044                         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1045                         ret = relayd_send_index(&relayd->control_sock, element,
1046                                 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1047                         if (ret < 0) {
1048                                 /*
1049                                  * Communication error with lttng-relayd,
1050                                  * perform cleanup now
1051                                  */
1052                                 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
1053                                 lttng_consumer_cleanup_relayd(relayd);
1054                                 ret = -1;
1055                         }
1056                         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1057                 } else {
1058                         ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
1059                                         stream->key, stream->net_seq_idx);
1060                         ret = -1;
1061                 }
1062         } else {
1063                 if (lttng_index_file_write(stream->index_file, element)) {
1064                         ret = -1;
1065                 } else {
1066                         ret = 0;
1067                 }
1068         }
1069         if (ret < 0) {
1070                 goto error;
1071         }
1072
1073 error:
1074         rcu_read_unlock();
1075         return ret;
1076 }
1077
1078 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
1079                 bool create_index)
1080 {
1081         int ret;
1082         enum lttng_trace_chunk_status chunk_status;
1083         const int flags = O_WRONLY | O_CREAT | O_TRUNC;
1084         const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
1085         char stream_path[LTTNG_PATH_MAX];
1086
1087         ASSERT_LOCKED(stream->lock);
1088         assert(stream->trace_chunk);
1089
1090         ret = utils_stream_file_path(stream->chan->pathname, stream->name,
1091                         stream->chan->tracefile_size,
1092                         stream->tracefile_count_current, NULL,
1093                         stream_path, sizeof(stream_path));
1094         if (ret < 0) {
1095                 goto end;
1096         }
1097
1098         if (stream->out_fd >= 0) {
1099                 ret = close(stream->out_fd);
1100                 if (ret < 0) {
1101                         PERROR("Failed to close stream file \"%s\"",
1102                                         stream->name);
1103                         goto end;
1104                 }
1105                 stream->out_fd = -1;
1106         }
1107
1108         DBG("Opening stream output file \"%s\"", stream_path);
1109         chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
1110                         flags, mode, &stream->out_fd, false);
1111         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1112                 ERR("Failed to open stream file \"%s\"", stream->name);
1113                 ret = -1;
1114                 goto end;
1115         }
1116
1117         if (!stream->metadata_flag && (create_index || stream->index_file)) {
1118                 if (stream->index_file) {
1119                         lttng_index_file_put(stream->index_file);
1120                 }
1121                 chunk_status = lttng_index_file_create_from_trace_chunk(
1122                                 stream->trace_chunk,
1123                                 stream->chan->pathname,
1124                                 stream->name,
1125                                 stream->chan->tracefile_size,
1126                                 stream->tracefile_count_current,
1127                                 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
1128                                 false, &stream->index_file);
1129                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1130                         ret = -1;
1131                         goto end;
1132                 }
1133         }
1134
1135         /* Reset current size because we just perform a rotation. */
1136         stream->tracefile_size_current = 0;
1137         stream->out_fd_offset = 0;
1138 end:
1139         return ret;
1140 }
1141
1142 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
1143 {
1144         int ret;
1145
1146         stream->tracefile_count_current++;
1147         if (stream->chan->tracefile_count > 0) {
1148                 stream->tracefile_count_current %=
1149                                 stream->chan->tracefile_count;
1150         }
1151
1152         DBG("Rotating output files of stream \"%s\"", stream->name);
1153         ret = consumer_stream_create_output_files(stream, true);
1154         if (ret) {
1155                 goto end;
1156         }
1157
1158 end:
1159         return ret;
1160 }
1161
1162 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
1163 {
1164         /*
1165          * This function does not take a const stream since
1166          * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1167          */
1168         assert(stream);
1169         return cds_lfht_is_node_deleted(&stream->node.node);
1170 }
1171
1172 static ssize_t metadata_bucket_flush(
1173                 const struct stream_subbuffer *buffer, void *data)
1174 {
1175         ssize_t ret;
1176         struct lttng_consumer_stream *stream = data;
1177
1178         ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1179         if (ret < 0) {
1180                 goto end;
1181         }
1182 end:
1183         return ret;
1184 }
1185
1186 static ssize_t metadata_bucket_consume(
1187                 struct lttng_consumer_local_data *unused,
1188                 struct lttng_consumer_stream *stream,
1189                 const struct stream_subbuffer *subbuffer)
1190 {
1191         ssize_t ret;
1192         enum metadata_bucket_status status;
1193
1194         status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1195         switch (status) {
1196         case METADATA_BUCKET_STATUS_OK:
1197                 /* Return consumed size. */
1198                 ret = subbuffer->buffer.buffer.size;
1199                 break;
1200         default:
1201                 ret = -1;
1202         }
1203
1204         return ret;
1205 }
1206
1207 int consumer_stream_enable_metadata_bucketization(
1208                 struct lttng_consumer_stream *stream)
1209 {
1210         int ret = 0;
1211
1212         assert(stream->metadata_flag);
1213         assert(!stream->metadata_bucket);
1214         assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1215
1216         stream->metadata_bucket = metadata_bucket_create(
1217                         metadata_bucket_flush, stream);
1218         if (!stream->metadata_bucket) {
1219                 ret = -1;
1220                 goto end;
1221         }
1222
1223         stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1224 end:
1225         return ret;
1226 }
1227
1228 void consumer_stream_metadata_set_version(
1229                 struct lttng_consumer_stream *stream, uint64_t new_version)
1230 {
1231         assert(new_version > stream->metadata_version);
1232         stream->metadata_version = new_version;
1233         stream->reset_metadata_flag = 1;
1234
1235         if (stream->metadata_bucket) {
1236                 metadata_bucket_reset(stream->metadata_bucket);
1237         }
1238 }
1239
1240 int consumer_stream_flush_buffer(struct lttng_consumer_stream *stream,
1241                 bool producer_active)
1242 {
1243         int ret = 0;
1244
1245         switch (consumer_data.type) {
1246         case LTTNG_CONSUMER_KERNEL:
1247                 if (producer_active) {
1248                         ret = kernctl_buffer_flush(stream->wait_fd);
1249                         if (ret < 0) {
1250                                 ERR("Failed to flush kernel stream");
1251                                 goto end;
1252                         }
1253                 } else {
1254                         ret = kernctl_buffer_flush_empty(stream->wait_fd);
1255                         if (ret < 0) {
1256                                 /*
1257                                  * Doing a buffer flush which does not take into
1258                                  * account empty packets. This is not perfect,
1259                                  * but required as a fall-back when
1260                                  * "flush_empty" is not implemented by
1261                                  * lttng-modules.
1262                                  */
1263                                 ret = kernctl_buffer_flush(stream->wait_fd);
1264                                 if (ret < 0) {
1265                                         ERR("Failed to flush kernel stream");
1266                                         goto end;
1267                                 }
1268                         }
1269                 }
1270                 break;
1271         case LTTNG_CONSUMER32_UST:
1272         case LTTNG_CONSUMER64_UST:
1273                 lttng_ustconsumer_flush_buffer(stream, (int) producer_active);
1274                 break;
1275         default:
1276                 ERR("Unknown consumer_data type");
1277                 abort();
1278         }
1279
1280 end:
1281         return ret;
1282 }