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