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
  26 #include "consumer-stream.h"
  27
  28 /*
  29  * RCU call to free stream. MUST only be used with call_rcu().
  30  */
  31 static void free_stream_rcu(struct rcu_head *head)
  32 {
  33         struct lttng_ht_node_u64 *node =
  34                 caa_container_of(head, struct lttng_ht_node_u64, head);
  35         struct lttng_consumer_stream *stream =
  36                 caa_container_of(node, struct lttng_consumer_stream, node);
  37
  38         pthread_mutex_destroy(&stream->lock);
  39         free(stream);
  40 }
  41
  42 static void consumer_stream_data_lock_all(struct lttng_consumer_stream *stream)
  43 {
  44         pthread_mutex_lock(&stream->chan->lock);
  45         pthread_mutex_lock(&stream->lock);
  46 }
  47
  48 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream *stream)
  49 {
  50         pthread_mutex_unlock(&stream->lock);
  51         pthread_mutex_unlock(&stream->chan->lock);
  52 }
  53
  54 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream *stream)
  55 {
  56         consumer_stream_data_lock_all(stream);
  57         pthread_mutex_lock(&stream->metadata_rdv_lock);
  58 }
  59
  60 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream *stream)
  61 {
  62         pthread_mutex_unlock(&stream->metadata_rdv_lock);
  63         consumer_stream_data_unlock_all(stream);
  64 }
  65
  66 /* Only used for data streams. */
  67 static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
  68                 const struct stream_subbuffer *subbuf)
  69 {
  70         int ret = 0;
  71         uint64_t sequence_number;
  72         const uint64_t discarded_events =
  73                         LTTNG_OPTIONAL_GET(subbuf->info.data.sequence_number);
  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
 201         assert(metadata);
 202         assert(metadata->metadata_flag);
 203         assert(ctx);
 204
 205         /*
 206          * In UST, since we have to write the metadata from the cache packet
 207          * by packet, we might need to start this procedure multiple times
 208          * until all the metadata from the cache has been extracted.
 209          */
 210         do {
 211                 /*
 212                  * Steps :
 213                  * - Lock the metadata stream
 214                  * - Check if metadata stream node was deleted before locking.
 215                  *   - if yes, release and return success
 216                  * - Check if new metadata is ready (flush + snapshot pos)
 217                  * - If nothing : release and return.
 218                  * - Lock the metadata_rdv_lock
 219                  * - Unlock the metadata stream
 220                  * - cond_wait on metadata_rdv to wait the wakeup from the
 221                  *   metadata thread
 222                  * - Unlock the metadata_rdv_lock
 223                  */
 224                 pthread_mutex_lock(&metadata->lock);
 225
 226                 /*
 227                  * There is a possibility that we were able to acquire a reference on the
 228                  * stream from the RCU hash table but between then and now, the node might
 229                  * have been deleted just before the lock is acquired. Thus, after locking,
 230                  * we make sure the metadata node has not been deleted which means that the
 231                  * buffers are closed.
 232                  *
 233                  * In that case, there is no need to sync the metadata hence returning a
 234                  * success return code.
 235                  */
 236                 ret = cds_lfht_is_node_deleted(&metadata->node.node);
 237                 if (ret) {
 238                         ret = 0;
 239                         goto end_unlock_mutex;
 240                 }
 241
 242                 switch (ctx->type) {
 243                 case LTTNG_CONSUMER_KERNEL:
 244                         /*
 245                          * Empty the metadata cache and flush the current stream.
 246                          */
 247                         ret = lttng_kconsumer_sync_metadata(metadata);
 248                         break;
 249                 case LTTNG_CONSUMER32_UST:
 250                 case LTTNG_CONSUMER64_UST:
 251                         /*
 252                          * Ask the sessiond if we have new metadata waiting and update the
 253                          * consumer metadata cache.
 254                          */
 255                         ret = lttng_ustconsumer_sync_metadata(ctx, metadata);
 256                         break;
 257                 default:
 258                         assert(0);
 259                         ret = -1;
 260                         break;
 261                 }
 262                 /*
 263                  * Error or no new metadata, we exit here.
 264                  */
 265                 if (ret <= 0 || ret == ENODATA) {
 266                         goto end_unlock_mutex;
 267                 }
 268
 269                 /*
 270                  * At this point, new metadata have been flushed, so we wait on the
 271                  * rendez-vous point for the metadata thread to wake us up when it
 272                  * finishes consuming the metadata and continue execution.
 273                  */
 274
 275                 pthread_mutex_lock(&metadata->metadata_rdv_lock);
 276
 277                 /*
 278                  * Release metadata stream lock so the metadata thread can process it.
 279                  */
 280                 pthread_mutex_unlock(&metadata->lock);
 281
 282                 /*
 283                  * Wait on the rendez-vous point. Once woken up, it means the metadata was
 284                  * consumed and thus synchronization is achieved.
 285                  */
 286                 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
 287                 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
 288         } while (ret == EAGAIN);
 289
 290         /* Success */
 291         return 0;
 292
 293 end_unlock_mutex:
 294         pthread_mutex_unlock(&metadata->lock);
 295         return ret;
 296 }
 297
 298 /*
 299  * Synchronize the metadata using a given session ID. A successful acquisition
 300  * of a metadata stream will trigger a request to the session daemon and a
 301  * snapshot so the metadata thread can consume it.
 302  *
 303  * This function call is a rendez-vous point between the metadata thread and
 304  * the data thread.
 305  *
 306  * Return 0 on success or else a negative value.
 307  */
 308 int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
 309                 uint64_t session_id)
 310 {
 311         int ret;
 312         struct lttng_consumer_stream *stream = NULL;
 313         struct lttng_ht_iter iter;
 314         struct lttng_ht *ht;
 315
 316         assert(ctx);
 317
 318         /* Ease our life a bit. */
 319         ht = consumer_data.stream_list_ht;
 320
 321         rcu_read_lock();
 322
 323         /* Search the metadata associated with the session id of the given stream. */
 324
 325         cds_lfht_for_each_entry_duplicate(ht->ht,
 326                         ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
 327                         &session_id, &iter.iter, stream, node_session_id.node) {
 328                 if (!stream->metadata_flag) {
 329                         continue;
 330                 }
 331
 332                 ret = do_sync_metadata(stream, ctx);
 333                 if (ret < 0) {
 334                         goto end;
 335                 }
 336         }
 337
 338         /*
 339          * Force return code to 0 (success) since ret might be ENODATA for instance
 340          * which is not an error but rather that we should come back.
 341          */
 342         ret = 0;
 343
 344 end:
 345         rcu_read_unlock();
 346         return ret;
 347 }
 348
 349 static int consumer_stream_sync_metadata_index(
 350                 struct lttng_consumer_stream *stream,
 351                 const struct stream_subbuffer *subbuffer,
 352                 struct lttng_consumer_local_data *ctx)
 353 {
 354         int ret;
 355
 356         /* Block until all the metadata is sent. */
 357         pthread_mutex_lock(&stream->metadata_timer_lock);
 358         assert(!stream->missed_metadata_flush);
 359         stream->waiting_on_metadata = true;
 360         pthread_mutex_unlock(&stream->metadata_timer_lock);
 361
 362         ret = consumer_stream_sync_metadata(ctx, stream->session_id);
 363
 364         pthread_mutex_lock(&stream->metadata_timer_lock);
 365         stream->waiting_on_metadata = false;
 366         if (stream->missed_metadata_flush) {
 367                 stream->missed_metadata_flush = false;
 368                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 369                 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
 370         } else {
 371                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 372         }
 373         if (ret < 0) {
 374                 goto end;
 375         }
 376
 377         ret = consumer_stream_send_index(stream, subbuffer, ctx);
 378 end:
 379         return ret;
 380 }
 381
 382 /*
 383  * Check if the local version of the metadata stream matches with the version
 384  * of the metadata stream in the kernel. If it was updated, set the reset flag
 385  * on the stream.
 386  */
 387 static
 388 int metadata_stream_check_version(struct lttng_consumer_stream *stream,
 389         const struct stream_subbuffer *subbuffer)
 390 {
 391         if (stream->metadata_version == subbuffer->info.metadata.version) {
 392                 goto end;
 393         }
 394
 395         DBG("New metadata version detected");
 396         consumer_stream_metadata_set_version(stream,
 397                         subbuffer->info.metadata.version);
 398
 399         if (stream->read_subbuffer_ops.reset_metadata) {
 400                 stream->read_subbuffer_ops.reset_metadata(stream);
 401         }
 402
 403 end:
 404         return 0;
 405 }
 406
 407 struct lttng_consumer_stream *consumer_stream_create(
 408                 struct lttng_consumer_channel *channel,
 409                 uint64_t channel_key,
 410                 uint64_t stream_key,
 411                 const char *channel_name,
 412                 uint64_t relayd_id,
 413                 uint64_t session_id,
 414                 struct lttng_trace_chunk *trace_chunk,
 415                 int cpu,
 416                 int *alloc_ret,
 417                 enum consumer_channel_type type,
 418                 unsigned int monitor)
 419 {
 420         int ret;
 421         struct lttng_consumer_stream *stream;
 422
 423         stream = zmalloc(sizeof(*stream));
 424         if (stream == NULL) {
 425                 PERROR("malloc struct lttng_consumer_stream");
 426                 ret = -ENOMEM;
 427                 goto end;
 428         }
 429
 430         if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
 431                 ERR("Failed to acquire trace chunk reference during the creation of a stream");
 432                 ret = -1;
 433                 goto error;
 434         }
 435
 436         rcu_read_lock();
 437         stream->chan = channel;
 438         stream->key = stream_key;
 439         stream->trace_chunk = trace_chunk;
 440         stream->out_fd = -1;
 441         stream->out_fd_offset = 0;
 442         stream->output_written = 0;
 443         stream->net_seq_idx = relayd_id;
 444         stream->session_id = session_id;
 445         stream->monitor = monitor;
 446         stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
 447         stream->index_file = NULL;
 448         stream->last_sequence_number = -1ULL;
 449         stream->rotate_position = -1ULL;
 450         pthread_mutex_init(&stream->lock, NULL);
 451         pthread_mutex_init(&stream->metadata_timer_lock, NULL);
 452
 453         /* If channel is the metadata, flag this stream as metadata. */
 454         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 455                 stream->metadata_flag = 1;
 456                 /* Metadata is flat out. */
 457                 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
 458                 /* Live rendez-vous point. */
 459                 pthread_cond_init(&stream->metadata_rdv, NULL);
 460                 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
 461         } else {
 462                 /* Format stream name to <channel_name>_<cpu_number> */
 463                 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
 464                                 channel_name, cpu);
 465                 if (ret < 0) {
 466                         PERROR("snprintf stream name");
 467                         goto error;
 468                 }
 469         }
 470
 471         switch (channel->output) {
 472         case CONSUMER_CHANNEL_SPLICE:
 473                 stream->output = LTTNG_EVENT_SPLICE;
 474                 ret = utils_create_pipe(stream->splice_pipe);
 475                 if (ret < 0) {
 476                         goto error;
 477                 }
 478                 break;
 479         case CONSUMER_CHANNEL_MMAP:
 480                 stream->output = LTTNG_EVENT_MMAP;
 481                 break;
 482         default:
 483                 abort();
 484         }
 485
 486         /* Key is always the wait_fd for streams. */
 487         lttng_ht_node_init_u64(&stream->node, stream->key);
 488
 489         /* Init node per channel id key */
 490         lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
 491
 492         /* Init session id node with the stream session id */
 493         lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
 494
 495         DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
 496                         " relayd_id %" PRIu64 ", session_id %" PRIu64,
 497                         stream->name, stream->key, channel_key,
 498                         stream->net_seq_idx, stream->session_id);
 499
 500         rcu_read_unlock();
 501
 502         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 503                 stream->read_subbuffer_ops.lock =
 504                                 consumer_stream_metadata_lock_all;
 505                 stream->read_subbuffer_ops.unlock =
 506                                 consumer_stream_metadata_unlock_all;
 507                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 508                                 metadata_stream_check_version;
 509         } else {
 510                 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
 511                 stream->read_subbuffer_ops.unlock =
 512                                 consumer_stream_data_unlock_all;
 513                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 514                                 consumer_stream_update_stats;
 515                 if (channel->is_live) {
 516                         stream->read_subbuffer_ops.post_consume =
 517                                         consumer_stream_sync_metadata_index;
 518                 } else {
 519                         stream->read_subbuffer_ops.post_consume =
 520                                         consumer_stream_send_index;
 521                 }
 522         }
 523
 524         if (channel->output == CONSUMER_CHANNEL_MMAP) {
 525                 stream->read_subbuffer_ops.consume_subbuffer =
 526                                 consumer_stream_consume_mmap;
 527         } else {
 528                 stream->read_subbuffer_ops.consume_subbuffer =
 529                                 consumer_stream_consume_splice;
 530         }
 531
 532         return stream;
 533
 534 error:
 535         rcu_read_unlock();
 536         lttng_trace_chunk_put(stream->trace_chunk);
 537         free(stream);
 538 end:
 539         if (alloc_ret) {
 540                 *alloc_ret = ret;
 541         }
 542         return NULL;
 543 }
 544
 545 /*
 546  * Close stream on the relayd side. This call can destroy a relayd if the
 547  * conditions are met.
 548  *
 549  * A RCU read side lock MUST be acquired if the relayd object was looked up in
 550  * a hash table before calling this.
 551  */
 552 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
 553                 struct consumer_relayd_sock_pair *relayd)
 554 {
 555         int ret;
 556
 557         assert(stream);
 558         assert(relayd);
 559
 560         if (stream->sent_to_relayd) {
 561                 uatomic_dec(&relayd->refcount);
 562                 assert(uatomic_read(&relayd->refcount) >= 0);
 563         }
 564
 565         /* Closing streams requires to lock the control socket. */
 566         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 567         ret = relayd_send_close_stream(&relayd->control_sock,
 568                         stream->relayd_stream_id,
 569                         stream->next_net_seq_num - 1);
 570         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 571         if (ret < 0) {
 572                 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
 573                 lttng_consumer_cleanup_relayd(relayd);
 574         }
 575
 576         /* Both conditions are met, we destroy the relayd. */
 577         if (uatomic_read(&relayd->refcount) == 0 &&
 578                         uatomic_read(&relayd->destroy_flag)) {
 579                 consumer_destroy_relayd(relayd);
 580         }
 581         stream->net_seq_idx = (uint64_t) -1ULL;
 582         stream->sent_to_relayd = 0;
 583 }
 584
 585 /*
 586  * Close stream's file descriptors and, if needed, close stream also on the
 587  * relayd side.
 588  *
 589  * The consumer data lock MUST be acquired.
 590  * The stream lock MUST be acquired.
 591  */
 592 void consumer_stream_close(struct lttng_consumer_stream *stream)
 593 {
 594         int ret;
 595         struct consumer_relayd_sock_pair *relayd;
 596
 597         assert(stream);
 598
 599         switch (consumer_data.type) {
 600         case LTTNG_CONSUMER_KERNEL:
 601                 if (stream->mmap_base != NULL) {
 602                         ret = munmap(stream->mmap_base, stream->mmap_len);
 603                         if (ret != 0) {
 604                                 PERROR("munmap");
 605                         }
 606                 }
 607
 608                 if (stream->wait_fd >= 0) {
 609                         ret = close(stream->wait_fd);
 610                         if (ret) {
 611                                 PERROR("close");
 612                         }
 613                         stream->wait_fd = -1;
 614                 }
 615                 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
 616                         utils_close_pipe(stream->splice_pipe);
 617                 }
 618                 break;
 619         case LTTNG_CONSUMER32_UST:
 620         case LTTNG_CONSUMER64_UST:
 621         {
 622                 /*
 623                  * Special case for the metadata since the wait fd is an internal pipe
 624                  * polled in the metadata thread.
 625                  */
 626                 if (stream->metadata_flag && stream->chan->monitor) {
 627                         int rpipe = stream->ust_metadata_poll_pipe[0];
 628
 629                         /*
 630                          * This will stop the channel timer if one and close the write side
 631                          * of the metadata poll pipe.
 632                          */
 633                         lttng_ustconsumer_close_metadata(stream->chan);
 634                         if (rpipe >= 0) {
 635                                 ret = close(rpipe);
 636                                 if (ret < 0) {
 637                                         PERROR("closing metadata pipe read side");
 638                                 }
 639                                 stream->ust_metadata_poll_pipe[0] = -1;
 640                         }
 641                 }
 642                 break;
 643         }
 644         default:
 645                 ERR("Unknown consumer_data type");
 646                 assert(0);
 647         }
 648
 649         /* Close output fd. Could be a socket or local file at this point. */
 650         if (stream->out_fd >= 0) {
 651                 ret = close(stream->out_fd);
 652                 if (ret) {
 653                         PERROR("close");
 654                 }
 655                 stream->out_fd = -1;
 656         }
 657
 658         if (stream->index_file) {
 659                 lttng_index_file_put(stream->index_file);
 660                 stream->index_file = NULL;
 661         }
 662
 663         lttng_trace_chunk_put(stream->trace_chunk);
 664         stream->trace_chunk = NULL;
 665
 666         /* Check and cleanup relayd if needed. */
 667         rcu_read_lock();
 668         relayd = consumer_find_relayd(stream->net_seq_idx);
 669         if (relayd != NULL) {
 670                 consumer_stream_relayd_close(stream, relayd);
 671         }
 672         rcu_read_unlock();
 673 }
 674
 675 /*
 676  * Delete the stream from all possible hash tables.
 677  *
 678  * The consumer data lock MUST be acquired.
 679  * The stream lock MUST be acquired.
 680  */
 681 void consumer_stream_delete(struct lttng_consumer_stream *stream,
 682                 struct lttng_ht *ht)
 683 {
 684         int ret;
 685         struct lttng_ht_iter iter;
 686
 687         assert(stream);
 688         /* Should NEVER be called not in monitor mode. */
 689         assert(stream->chan->monitor);
 690
 691         rcu_read_lock();
 692
 693         if (ht) {
 694                 iter.iter.node = &stream->node.node;
 695                 ret = lttng_ht_del(ht, &iter);
 696                 assert(!ret);
 697         }
 698
 699         /* Delete from stream per channel ID hash table. */
 700         iter.iter.node = &stream->node_channel_id.node;
 701         /*
 702          * The returned value is of no importance. Even if the node is NOT in the
 703          * hash table, we continue since we may have been called by a code path
 704          * that did not add the stream to a (all) hash table. Same goes for the
 705          * next call ht del call.
 706          */
 707         (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
 708
 709         /* Delete from the global stream list. */
 710         iter.iter.node = &stream->node_session_id.node;
 711         /* See the previous ht del on why we ignore the returned value. */
 712         (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
 713
 714         rcu_read_unlock();
 715
 716         if (!stream->metadata_flag) {
 717                 /* Decrement the stream count of the global consumer data. */
 718                 assert(consumer_data.stream_count > 0);
 719                 consumer_data.stream_count--;
 720         }
 721 }
 722
 723 /*
 724  * Free the given stream within a RCU call.
 725  */
 726 void consumer_stream_free(struct lttng_consumer_stream *stream)
 727 {
 728         assert(stream);
 729
 730         metadata_bucket_destroy(stream->metadata_bucket);
 731         call_rcu(&stream->node.head, free_stream_rcu);
 732 }
 733
 734 /*
 735  * Destroy the stream's buffers of the tracer.
 736  */
 737 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 738 {
 739         assert(stream);
 740
 741         switch (consumer_data.type) {
 742         case LTTNG_CONSUMER_KERNEL:
 743                 break;
 744         case LTTNG_CONSUMER32_UST:
 745         case LTTNG_CONSUMER64_UST:
 746                 lttng_ustconsumer_del_stream(stream);
 747                 break;
 748         default:
 749                 ERR("Unknown consumer_data type");
 750                 assert(0);
 751         }
 752 }
 753
 754 /*
 755  * Destroy and close a already created stream.
 756  */
 757 static void destroy_close_stream(struct lttng_consumer_stream *stream)
 758 {
 759         assert(stream);
 760
 761         DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
 762
 763         /* Destroy tracer buffers of the stream. */
 764         consumer_stream_destroy_buffers(stream);
 765         /* Close down everything including the relayd if one. */
 766         consumer_stream_close(stream);
 767 }
 768
 769 /*
 770  * Decrement the stream's channel refcount and if down to 0, return the channel
 771  * pointer so it can be destroyed by the caller or NULL if not.
 772  */
 773 static struct lttng_consumer_channel *unref_channel(
 774                 struct lttng_consumer_stream *stream)
 775 {
 776         struct lttng_consumer_channel *free_chan = NULL;
 777
 778         assert(stream);
 779         assert(stream->chan);
 780
 781         /* Update refcount of channel and see if we need to destroy it. */
 782         if (!uatomic_sub_return(&stream->chan->refcount, 1)
 783                         && !uatomic_read(&stream->chan->nb_init_stream_left)) {
 784                 free_chan = stream->chan;
 785         }
 786
 787         return free_chan;
 788 }
 789
 790 /*
 791  * Destroy a stream completely. This will delete, close and free the stream.
 792  * Once return, the stream is NO longer usable. Its channel may get destroyed
 793  * if conditions are met for a monitored stream.
 794  *
 795  * This MUST be called WITHOUT the consumer data and stream lock acquired if
 796  * the stream is in _monitor_ mode else it does not matter.
 797  */
 798 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 799                 struct lttng_ht *ht)
 800 {
 801         assert(stream);
 802
 803         /* Stream is in monitor mode. */
 804         if (stream->monitor) {
 805                 struct lttng_consumer_channel *free_chan = NULL;
 806
 807                 /*
 808                  * This means that the stream was successfully removed from the streams
 809                  * list of the channel and sent to the right thread managing this
 810                  * stream thus being globally visible.
 811                  */
 812                 if (stream->globally_visible) {
 813                         pthread_mutex_lock(&consumer_data.lock);
 814                         pthread_mutex_lock(&stream->chan->lock);
 815                         pthread_mutex_lock(&stream->lock);
 816                         /* Remove every reference of the stream in the consumer. */
 817                         consumer_stream_delete(stream, ht);
 818
 819                         destroy_close_stream(stream);
 820
 821                         /* Update channel's refcount of the stream. */
 822                         free_chan = unref_channel(stream);
 823
 824                         /* Indicates that the consumer data state MUST be updated after this. */
 825                         consumer_data.need_update = 1;
 826
 827                         pthread_mutex_unlock(&stream->lock);
 828                         pthread_mutex_unlock(&stream->chan->lock);
 829                         pthread_mutex_unlock(&consumer_data.lock);
 830                 } else {
 831                         /*
 832                          * If the stream is not visible globally, this needs to be done
 833                          * outside of the consumer data lock section.
 834                          */
 835                         free_chan = unref_channel(stream);
 836                 }
 837
 838                 if (free_chan) {
 839                         consumer_del_channel(free_chan);
 840                 }
 841         } else {
 842                 destroy_close_stream(stream);
 843         }
 844
 845         /* Free stream within a RCU call. */
 846         lttng_trace_chunk_put(stream->trace_chunk);
 847         stream->trace_chunk = NULL;
 848         consumer_stream_free(stream);
 849 }
 850
 851 /*
 852  * Write index of a specific stream either on the relayd or local disk.
 853  *
 854  * Return 0 on success or else a negative value.
 855  */
 856 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
 857                 struct ctf_packet_index *element)
 858 {
 859         int ret;
 860
 861         assert(stream);
 862         assert(element);
 863
 864         rcu_read_lock();
 865         if (stream->net_seq_idx != (uint64_t) -1ULL) {
 866                 struct consumer_relayd_sock_pair *relayd;
 867                 relayd = consumer_find_relayd(stream->net_seq_idx);
 868                 if (relayd) {
 869                         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 870                         ret = relayd_send_index(&relayd->control_sock, element,
 871                                 stream->relayd_stream_id, stream->next_net_seq_num - 1);
 872                         if (ret < 0) {
 873                                 /*
 874                                  * Communication error with lttng-relayd,
 875                                  * perform cleanup now
 876                                  */
 877                                 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
 878                                 lttng_consumer_cleanup_relayd(relayd);
 879                                 ret = -1;
 880                         }
 881                         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 882                 } else {
 883                         ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
 884                                         stream->key, stream->net_seq_idx);
 885                         ret = -1;
 886                 }
 887         } else {
 888                 if (lttng_index_file_write(stream->index_file, element)) {
 889                         ret = -1;
 890                 } else {
 891                         ret = 0;
 892                 }
 893         }
 894         if (ret < 0) {
 895                 goto error;
 896         }
 897
 898 error:
 899         rcu_read_unlock();
 900         return ret;
 901 }
 902
 903 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
 904                 bool create_index)
 905 {
 906         int ret;
 907         enum lttng_trace_chunk_status chunk_status;
 908         const int flags = O_WRONLY | O_CREAT | O_TRUNC;
 909         const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
 910         char stream_path[LTTNG_PATH_MAX];
 911
 912         ASSERT_LOCKED(stream->lock);
 913         assert(stream->trace_chunk);
 914
 915         ret = utils_stream_file_path(stream->chan->pathname, stream->name,
 916                         stream->chan->tracefile_size,
 917                         stream->tracefile_count_current, NULL,
 918                         stream_path, sizeof(stream_path));
 919         if (ret < 0) {
 920                 goto end;
 921         }
 922
 923         if (stream->out_fd >= 0) {
 924                 ret = close(stream->out_fd);
 925                 if (ret < 0) {
 926                         PERROR("Failed to close stream file \"%s\"",
 927                                         stream->name);
 928                         goto end;
 929                 }
 930                 stream->out_fd = -1;
 931         }
 932
 933         DBG("Opening stream output file \"%s\"", stream_path);
 934         chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
 935                         flags, mode, &stream->out_fd, false);
 936         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 937                 ERR("Failed to open stream file \"%s\"", stream->name);
 938                 ret = -1;
 939                 goto end;
 940         }
 941
 942         if (!stream->metadata_flag && (create_index || stream->index_file)) {
 943                 if (stream->index_file) {
 944                         lttng_index_file_put(stream->index_file);
 945                 }
 946                 chunk_status = lttng_index_file_create_from_trace_chunk(
 947                                 stream->trace_chunk,
 948                                 stream->chan->pathname,
 949                                 stream->name,
 950                                 stream->chan->tracefile_size,
 951                                 stream->tracefile_count_current,
 952                                 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
 953                                 false, &stream->index_file);
 954                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 955                         ret = -1;
 956                         goto end;
 957                 }
 958         }
 959
 960         /* Reset current size because we just perform a rotation. */
 961         stream->tracefile_size_current = 0;
 962         stream->out_fd_offset = 0;
 963 end:
 964         return ret;
 965 }
 966
 967 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
 968 {
 969         int ret;
 970
 971         stream->tracefile_count_current++;
 972         if (stream->chan->tracefile_count > 0) {
 973                 stream->tracefile_count_current %=
 974                                 stream->chan->tracefile_count;
 975         }
 976
 977         DBG("Rotating output files of stream \"%s\"", stream->name);
 978         ret = consumer_stream_create_output_files(stream, true);
 979         if (ret) {
 980                 goto end;
 981         }
 982
 983 end:
 984         return ret;
 985 }
 986
 987 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
 988 {
 989         /*
 990          * This function does not take a const stream since
 991          * cds_lfht_is_node_deleted was not const before liburcu 0.12.
 992          */
 993         assert(stream);
 994         return cds_lfht_is_node_deleted(&stream->node.node);
 995 }
 996
 997 static ssize_t metadata_bucket_flush(
 998                 const struct stream_subbuffer *buffer, void *data)
 999 {
1000         ssize_t ret;
1001         struct lttng_consumer_stream *stream = data;
1002
1003         ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1004         if (ret < 0) {
1005                 goto end;
1006         }
1007 end:
1008         return ret;
1009 }
1010
1011 static ssize_t metadata_bucket_consume(
1012                 struct lttng_consumer_local_data *unused,
1013                 struct lttng_consumer_stream *stream,
1014                 const struct stream_subbuffer *subbuffer)
1015 {
1016         ssize_t ret;
1017         enum metadata_bucket_status status;
1018
1019         status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1020         switch (status) {
1021         case METADATA_BUCKET_STATUS_OK:
1022                 /* Return consumed size. */
1023                 ret = subbuffer->buffer.buffer.size;
1024                 break;
1025         default:
1026                 ret = -1;
1027         }
1028
1029         return ret;
1030 }
1031
1032 int consumer_stream_enable_metadata_bucketization(
1033                 struct lttng_consumer_stream *stream)
1034 {
1035         int ret = 0;
1036
1037         assert(stream->metadata_flag);
1038         assert(!stream->metadata_bucket);
1039         assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1040
1041         stream->metadata_bucket = metadata_bucket_create(
1042                         metadata_bucket_flush, stream);
1043         if (!stream->metadata_bucket) {
1044                 ret = -1;
1045                 goto end;
1046         }
1047
1048         stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1049 end:
1050         return ret;
1051 }
1052
1053 void consumer_stream_metadata_set_version(
1054                 struct lttng_consumer_stream *stream, uint64_t new_version)
1055 {
1056         assert(new_version > stream->metadata_version);
1057         stream->metadata_version = new_version;
1058         stream->reset_metadata_flag = 1;
1059
1060         if (stream->metadata_bucket) {
1061                 metadata_bucket_reset(stream->metadata_bucket);
1062         }
1063 }