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